Skip to content
Projects
Groups
Snippets
Help
Loading...
Help
Support
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in / Register
Toggle navigation
P
pistache
Project overview
Project overview
Details
Activity
Releases
Repository
Repository
Files
Commits
Branches
Tags
Contributors
Graph
Compare
Issues
0
Issues
0
List
Boards
Labels
Milestones
Merge Requests
0
Merge Requests
0
CI / CD
CI / CD
Pipelines
Jobs
Schedules
Analytics
Analytics
CI / CD
Repository
Value Stream
Wiki
Wiki
Snippets
Snippets
Members
Members
Collapse sidebar
Close sidebar
Activity
Graph
Create a new issue
Jobs
Commits
Issue Boards
Open sidebar
Libraries
pistache
Commits
9fcc1835
Commit
9fcc1835
authored
Aug 14, 2015
by
Mathieu Stefani
Committed by
Mathieu STEFANI
Aug 23, 2015
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Added a basic HTTP handler and request / response classes
parent
17282a44
Changes
13
Hide whitespace changes
Inline
Side-by-side
Showing
13 changed files
with
496 additions
and
93 deletions
+496
-93
main.cc
main.cc
+18
-0
src/.CMakeLists.txt.swp
src/.CMakeLists.txt.swp
+0
-0
src/CMakeLists.txt
src/CMakeLists.txt
+3
-0
src/common.h
src/common.h
+5
-2
src/http.cc
src/http.cc
+139
-25
src/http.h
src/http.h
+80
-10
src/listener.cc
src/listener.cc
+140
-28
src/listener.h
src/listener.h
+50
-14
src/mailbox.h
src/mailbox.h
+4
-0
src/net.cc
src/net.cc
+16
-11
src/net.h
src/net.h
+11
-2
src/peer.cc
src/peer.cc
+21
-0
src/peer.h
src/peer.h
+9
-1
No files found.
main.cc
View file @
9fcc1835
#include "net.h"
#include "net.h"
#include "peer.h"
#include "http.h"
#include "http.h"
#include <iostream>
#include <iostream>
#include <cstring>
#include <cstring>
using
namespace
std
;
using
namespace
std
;
class
MyHandler
:
public
Net
::
Http
::
Handler
{
void
onRequest
(
const
Net
::
Http
::
Request
&
req
,
Net
::
Tcp
::
Peer
&
peer
)
{
cout
<<
"Received "
<<
methodString
(
req
.
method
)
<<
" request on "
<<
req
.
resource
<<
endl
;
if
(
req
.
resource
==
"/ping"
)
{
if
(
req
.
method
==
Net
::
Http
::
Method
::
Get
)
{
cout
<<
"PONG"
<<
endl
;
Net
::
Http
::
Response
response
(
Net
::
Http
::
Code
::
Ok
,
"PONG"
);
response
.
writeTo
(
peer
);
}
}
}
};
int
main
(
int
argc
,
char
*
argv
[])
{
int
main
(
int
argc
,
char
*
argv
[])
{
Net
::
Port
port
(
9080
);
Net
::
Port
port
(
9080
);
...
@@ -18,5 +34,7 @@ int main(int argc, char *argv[]) {
...
@@ -18,5 +34,7 @@ int main(int argc, char *argv[]) {
cout
<<
"Cores = "
<<
hardware_concurrency
()
<<
endl
;
cout
<<
"Cores = "
<<
hardware_concurrency
()
<<
endl
;
Net
::
Http
::
Server
server
(
addr
);
Net
::
Http
::
Server
server
(
addr
);
server
.
setHandler
(
std
::
make_shared
<
MyHandler
>
());
server
.
serve
();
server
.
serve
();
}
}
src/.CMakeLists.txt.swp
deleted
100644 → 0
View file @
17282a44
File deleted
src/CMakeLists.txt
View file @
9fcc1835
...
@@ -2,6 +2,9 @@ set(SOURCE_FILES
...
@@ -2,6 +2,9 @@ set(SOURCE_FILES
listener.cc
listener.cc
net.cc
net.cc
peer.cc
peer.cc
os.cc
peer.cc
http.cc
)
)
add_library
(
net
${
SOURCE_FILES
}
)
add_library
(
net
${
SOURCE_FILES
}
)
...
...
src/common.h
View file @
9fcc1835
...
@@ -6,6 +6,7 @@
...
@@ -6,6 +6,7 @@
#pragma once
#pragma once
#include <sstream>
#include <cstdio>
#include <cstdio>
#include <cassert>
#include <cassert>
#include <sys/types.h>
#include <sys/types.h>
...
@@ -26,8 +27,10 @@
...
@@ -26,8 +27,10 @@
[&]() { \
[&]() { \
auto ret = __VA_ARGS__; \
auto ret = __VA_ARGS__; \
if (ret < 0) { \
if (ret < 0) { \
perror(#__VA_ARGS__); \
const char *str = #__VA_ARGS__; \
throw std::runtime_error(#__VA_ARGS__); \
std::ostringstream oss; \
oss << str << ": " << strerror(errno); \
throw std::runtime_error(oss.str()); \
} \
} \
return ret; \
return ret; \
}(); \
}(); \
...
...
src/http.cc
View file @
9fcc1835
...
@@ -7,8 +7,10 @@
...
@@ -7,8 +7,10 @@
#include <cstring>
#include <cstring>
#include <iostream>
#include <iostream>
#include <stdexcept>
#include <stdexcept>
#include "common.h"
#include "http.h"
#include "http.h"
#include "net.h"
#include "net.h"
#include "peer.h"
using
namespace
std
;
using
namespace
std
;
...
@@ -16,10 +18,11 @@ namespace Net {
...
@@ -16,10 +18,11 @@ namespace Net {
namespace
Http
{
namespace
Http
{
namespace
Private
{
static
constexpr
char
CR
=
0xD
;
static
constexpr
char
LF
=
0xA
;
static
constexpr
char
CRLF
[]
=
{
CR
,
LF
};
static
constexpr
char
CR
=
0xD
;
namespace
Private
{
static
constexpr
char
LF
=
0xA
;
void
void
Parser
::
advance
(
size_t
count
)
{
Parser
::
advance
(
size_t
count
)
{
...
@@ -32,7 +35,25 @@ namespace Private {
...
@@ -32,7 +35,25 @@ namespace Private {
bool
bool
Parser
::
eol
()
const
{
Parser
::
eol
()
const
{
return
buffer
[
cursor
]
==
CR
&&
buffer
[
cursor
+
1
]
==
LF
;
return
buffer
[
cursor
]
==
CR
&&
next
()
==
LF
;
}
char
Parser
::
next
()
const
{
if
(
cursor
+
1
>=
len
)
{
throw
std
::
runtime_error
(
"Early EOF"
);
}
return
buffer
[
cursor
+
1
];
}
Request
Parser
::
expectRequest
()
{
expectRequestLine
();
expectHeaders
();
expectBody
();
return
request
;
}
}
// 5.1 Request-Line
// 5.1 Request-Line
...
@@ -99,6 +120,7 @@ namespace Private {
...
@@ -99,6 +120,7 @@ namespace Private {
void
void
Parser
::
expectHeaders
()
{
Parser
::
expectHeaders
()
{
while
(
!
eol
())
{
while
(
!
eol
())
{
// Read the header name
size_t
start
=
cursor
;
size_t
start
=
cursor
;
while
(
buffer
[
cursor
]
!=
':'
)
while
(
buffer
[
cursor
]
!=
':'
)
...
@@ -111,51 +133,135 @@ namespace Private {
...
@@ -111,51 +133,135 @@ namespace Private {
// Skip the ':'
// Skip the ':'
advance
(
1
);
advance
(
1
);
// Read the header value
start
=
cursor
;
start
=
cursor
;
while
(
!
eol
())
while
(
!
eol
())
advance
(
1
);
advance
(
1
);
std
::
string
fieldValue
=
std
::
string
(
buffer
+
start
,
cursor
-
start
);
std
::
string
fieldValue
=
std
::
string
(
buffer
+
start
,
cursor
-
start
);
advance
(
2
);
if
(
fieldName
==
"Content-Length"
)
{
size_t
pos
;
contentLength
=
std
::
stol
(
fieldValue
,
&
pos
);
}
request
.
headers
.
push_back
(
make_pair
(
std
::
move
(
fieldName
),
std
::
move
(
fieldValue
)));
request
.
headers
.
push_back
(
make_pair
(
std
::
move
(
fieldName
),
std
::
move
(
fieldValue
)));
// CRLF
advance
(
2
);
}
}
void
Parser
::
expectBody
()
{
if
(
contentLength
>
0
)
{
advance
(
2
);
// CRLF
if
(
cursor
+
contentLength
>
len
)
{
throw
std
::
runtime_error
(
"Corrupted HTTP Body"
);
}
request
.
body
=
std
::
string
(
buffer
+
cursor
,
contentLength
);
}
}
}
}
}
// namespace Private
}
// namespace Private
const
char
*
methodString
(
Method
method
)
{
switch
(
method
)
{
#define METHOD(name, str) \
case Method::name: \
return str;
HTTP_METHODS
#undef METHOD
}
return
nullptr
;
}
const
char
*
codeString
(
Code
code
)
{
switch
(
code
)
{
#define CODE(_, name, str) \
case Code::name: \
return str;
STATUS_CODES
#undef CODE
}
return
nullptr
;
}
Message
::
Message
()
{
}
Request
::
Request
()
Request
::
Request
()
:
Message
()
{
}
{
}
Response
::
Response
(
Code
code
,
std
::
string
body
)
:
Message
()
{
this
->
body
=
std
::
move
(
body
);
code_
=
code
;
}
const
char
*
methodString
(
Method
method
)
void
Response
::
writeTo
(
Tcp
::
Peer
&
peer
)
{
{
static
constexpr
size_t
MethodsCount
int
fd
=
peer
.
fd
();
=
sizeof
priv__methodStrings
/
sizeof
*
priv__methodStrings
;
char
buffer
[
Const
::
MaxBuffer
];
std
::
memset
(
buffer
,
0
,
Const
::
MaxBuffer
);
char
*
p_buf
=
buffer
;
int
index
=
static_cast
<
int
>
(
method
);
auto
writeRaw
=
[
&
](
const
void
*
data
,
size_t
len
)
{
if
(
index
>=
MethodsCount
)
p_buf
=
static_cast
<
char
*>
(
memcpy
(
p_buf
,
data
,
len
));
throw
std
::
logic_error
(
"Invalid method index"
);
p_buf
+=
len
;
};
return
priv__methodStrings
[
index
];
auto
writeString
=
[
&
](
const
char
*
str
)
{
const
size_t
len
=
std
::
strlen
(
str
);
writeRaw
(
str
,
std
::
strlen
(
str
));
};
auto
writeInt
=
[
&
](
uint64_t
value
)
{
auto
str
=
std
::
to_string
(
value
);
writeRaw
(
str
.
c_str
(),
str
.
size
());
};
auto
writeChar
=
[
&
](
char
c
)
{
*
p_buf
++
=
c
;
};
writeString
(
"HTTP/1.1 "
);
writeInt
(
static_cast
<
int
>
(
code_
));
writeChar
(
' '
);
writeString
(
codeString
(
code_
));
writeRaw
(
CRLF
,
2
);
writeString
(
"Content-Length:"
);
writeInt
(
body
.
size
());
writeRaw
(
CRLF
,
2
);
writeRaw
(
CRLF
,
2
);
writeString
(
body
.
c_str
());
const
size_t
len
=
p_buf
-
buffer
;
ssize_t
bytes
=
send
(
fd
,
buffer
,
len
,
0
);
cout
<<
bytes
<<
" bytes sent"
<<
endl
;
}
}
void
void
Handler
::
onInput
(
const
char
*
buffer
,
size_t
len
)
{
Handler
::
onInput
(
const
char
*
buffer
,
size_t
len
,
Tcp
::
Peer
&
peer
)
{
cout
<<
"Received http request "
<<
string
(
buffer
,
len
)
<<
endl
;
Private
::
Parser
parser
(
buffer
,
len
);
Private
::
Parser
parser
(
buffer
,
len
);
parser
.
expectRequestLine
();
auto
request
=
parser
.
expectRequest
();
parser
.
expectHeaders
();
auto
req
=
parser
.
request
;
onRequest
(
request
,
peer
);
cout
<<
"method = "
<<
methodString
(
req
.
method
)
<<
endl
;
cout
<<
"resource = "
<<
req
.
resource
<<
endl
;
cout
<<
"headers = "
<<
endl
;
for
(
const
auto
&
header
:
req
.
headers
)
{
cout
<<
string
(
' '
,
4
)
<<
header
.
first
<<
" -> "
<<
header
.
second
<<
endl
;
}
}
}
void
void
...
@@ -169,11 +275,19 @@ Server::Server(const Net::Address& addr)
...
@@ -169,11 +275,19 @@ Server::Server(const Net::Address& addr)
:
listener
(
addr
)
:
listener
(
addr
)
{
}
{
}
void
Server
::
setHandler
(
const
std
::
shared_ptr
<
Handler
>&
handler
)
{
handler_
=
handler
;
}
void
void
Server
::
serve
()
Server
::
serve
()
{
{
if
(
!
handler_
)
throw
std
::
runtime_error
(
"Must call setHandler() prior to serve()"
);
listener
.
init
(
4
);
listener
.
init
(
4
);
listener
.
setHandler
(
make_shared
<
Handler
>
()
);
listener
.
setHandler
(
handler_
);
if
(
listener
.
bind
())
{
if
(
listener
.
bind
())
{
const
auto
&
addr
=
listener
.
address
();
const
auto
&
addr
=
listener
.
address
();
...
...
src/http.h
View file @
9fcc1835
...
@@ -23,6 +23,48 @@ namespace Http {
...
@@ -23,6 +23,48 @@ namespace Http {
METHOD(Trace, "TRACE") \
METHOD(Trace, "TRACE") \
METHOD(Connect, "CONNECT")
METHOD(Connect, "CONNECT")
// 10. Status Code Definitions
#define STATUS_CODES \
CODE(100, Continue, "Continue") \
CODE(101, Switching_Protocols, "Switching Protocols") \
CODE(200, Ok, "OK") \
CODE(201, Created, "Created") \
CODE(202, Accepted, "Accepted") \
CODE(203, NonAuthoritative_Information, "Non-Authoritative Information") \
CODE(204, No_Content, "No Content") \
CODE(205, Reset_Content, "Reset Content") \
CODE(206, Partial_Content, "Partial Content") \
CODE(300, Multiple_Choices, "Multiple Choices") \
CODE(301, Moved_Permanently, "Moved Permanently") \
CODE(302, Found, "Found") \
CODE(303, See_Other, "See Other") \
CODE(304, Not_Modified, "Not Modified") \
CODE(305, Use_Proxy, "Use Proxy") \
CODE(307, Temporary_Redirect, "Temporary Redirect") \
CODE(400, Bad_Request, "Bad Request") \
CODE(401, Unauthorized, "Unauthorized") \
CODE(402, Payment_Required, "Payment Required") \
CODE(403, Forbidden, "Forbidden") \
CODE(404, Not_Found, "Not Found") \
CODE(405, Method_Not_Allowed, "Method Not Allowed") \
CODE(406, Not_Acceptable, "Not Acceptable") \
CODE(407, Proxy_Authentication_Required, "Proxy Authentication Required") \
CODE(408, Request_Timeout, "Request Timeout") \
CODE(409, Conflict, "Conflict") \
CODE(410, Gone, "Gone") \
CODE(411, Length_Required, "Length Required") \
CODE(412, Precondition_Failed, "Precondition Failed") \
CODE(413, Request_Entity_Too_Large, "Request Entity Too Large") \
CODE(414, RequestURI_Too_Long, "Request-URI Too Long") \
CODE(415, Unsupported_Media_Type, "Unsupported Media Type") \
CODE(416, Requested_Range_Not_Satisfiable, "Requested Range Not Satisfiable") \
CODE(417, Expectation_Failed, "Expectation Failed") \
CODE(500, Internal_Server_Error, "Internal Server Error") \
CODE(501, Not_Implemented, "Not Implemented") \
CODE(502, Bad_Gateway, "Bad Gateway") \
CODE(503, Service_Unavailable, "Service Unavailable") \
CODE(504, Gateway_Timeout, "Gateway Timeout")
enum
class
Method
{
enum
class
Method
{
#define METHOD(m, _) m,
#define METHOD(m, _) m,
...
@@ -30,23 +72,42 @@ enum class Method {
...
@@ -30,23 +72,42 @@ enum class Method {
#undef METHOD
#undef METHOD
};
};
static
constexpr
const
char
*
priv__methodStrings
[]
=
{
enum
class
Code
{
#define METHOD(_, str) #str,
#define CODE(value, name, _) name = value,
HTTP_METHODS
STATUS_CODES
#undef METHOD
#undef CODE
};
const
char
*
methodString
(
Method
method
);
const
char
*
codeString
(
Code
code
);
// 4. HTTP Message
class
Message
{
public:
Message
();
std
::
vector
<
std
::
pair
<
std
::
string
,
std
::
string
>>
headers
;
std
::
string
body
;
};
};
class
Request
{
// 5. Request
class
Request
:
public
Message
{
public:
public:
Request
();
Request
();
Method
method
;
Method
method
;
std
::
string
resource
;
std
::
string
resource
;
std
::
string
body
;
std
::
vector
<
std
::
pair
<
std
::
string
,
std
::
string
>>
headers
;
};
};
// 6. Response
class
Response
:
public
Message
{
public:
Response
(
Code
code
,
std
::
string
body
);
void
writeTo
(
Tcp
::
Peer
&
peer
);
private:
Code
code_
;
};
namespace
Private
{
namespace
Private
{
...
@@ -55,10 +116,13 @@ namespace Private {
...
@@ -55,10 +116,13 @@ namespace Private {
:
buffer
(
buffer
)
:
buffer
(
buffer
)
,
len
(
len
)
,
len
(
len
)
,
cursor
(
0
)
,
cursor
(
0
)
,
contentLength
(
-
1
)
{
}
{
}
Request
expectRequest
();
void
expectRequestLine
();
void
expectRequestLine
();
void
expectHeaders
();
void
expectHeaders
();
void
expectBody
();
void
advance
(
size_t
count
);
void
advance
(
size_t
count
);
bool
eol
()
const
;
bool
eol
()
const
;
...
@@ -67,7 +131,9 @@ namespace Private {
...
@@ -67,7 +131,9 @@ namespace Private {
size_t
len
;
size_t
len
;
size_t
cursor
;
size_t
cursor
;
char
next
()
const
;
private:
ssize_t
contentLength
;
Request
request
;
Request
request
;
};
};
...
@@ -75,8 +141,10 @@ namespace Private {
...
@@ -75,8 +141,10 @@ namespace Private {
class
Handler
:
public
Net
::
Tcp
::
Handler
{
class
Handler
:
public
Net
::
Tcp
::
Handler
{
public:
public:
void
onInput
(
const
char
*
buffer
,
size_t
len
);
void
onInput
(
const
char
*
buffer
,
size_t
len
,
Tcp
::
Peer
&
peer
);
void
onOutput
();
void
onOutput
();
virtual
void
onRequest
(
const
Request
&
request
,
Tcp
::
Peer
&
peer
)
=
0
;
};
};
class
Server
{
class
Server
{
...
@@ -84,9 +152,11 @@ public:
...
@@ -84,9 +152,11 @@ public:
Server
();
Server
();
Server
(
const
Net
::
Address
&
addr
);
Server
(
const
Net
::
Address
&
addr
);
void
setHandler
(
const
std
::
shared_ptr
<
Handler
>&
handler
);
void
serve
();
void
serve
();
private:
private:
std
::
shared_ptr
<
Handler
>
handler_
;
Net
::
Tcp
::
Listener
listener
;
Net
::
Tcp
::
Listener
listener
;
};
};
...
...
src/listener.cc
View file @
9fcc1835
...
@@ -16,6 +16,7 @@
...
@@ -16,6 +16,7 @@
#include "listener.h"
#include "listener.h"
#include "peer.h"
#include "peer.h"
#include "common.h"
#include "common.h"
#include "os.h"
using
namespace
std
;
using
namespace
std
;
...
@@ -23,6 +24,33 @@ namespace Net {
...
@@ -23,6 +24,33 @@ namespace Net {
namespace
Tcp
{
namespace
Tcp
{
struct
Message
{
virtual
~
Message
()
{
}
enum
class
Type
{
NewPeer
,
Shutdown
};
virtual
Type
type
()
const
=
0
;
};
struct
PeerMessage
:
public
Message
{
PeerMessage
(
const
std
::
shared_ptr
<
Peer
>&
peer
)
:
peer_
(
peer
)
{
}
Type
type
()
const
{
return
Type
::
NewPeer
;
}
std
::
shared_ptr
<
Peer
>
peer
()
const
{
return
peer_
;
}
private:
std
::
shared_ptr
<
Peer
>
peer_
;
};
template
<
typename
To
>
To
*
message_cast
(
const
std
::
unique_ptr
<
Message
>&
from
)
{
return
static_cast
<
To
*>
(
from
.
get
());
}
IoWorker
::
IoWorker
()
{
IoWorker
::
IoWorker
()
{
epoll_fd
=
TRY_RET
(
epoll_create
(
128
));
epoll_fd
=
TRY_RET
(
epoll_create
(
128
));
}
}
...
@@ -34,20 +62,33 @@ IoWorker::~IoWorker() {
...
@@ -34,20 +62,33 @@ IoWorker::~IoWorker() {
}
}
void
void
IoWorker
::
start
()
{
IoWorker
::
start
(
const
std
::
shared_ptr
<
Handler
>&
handler
)
{
handler_
=
handler
;
thread
.
reset
(
new
std
::
thread
([
this
]()
{
thread
.
reset
(
new
std
::
thread
([
this
]()
{
this
->
run
();
this
->
run
();
}));
}));
}
}
std
::
shared_ptr
<
Peer
>
IoWorker
::
getPeer
(
Fd
fd
)
const
{
auto
it
=
peers
.
find
(
fd
);
if
(
it
==
std
::
end
(
peers
))
{
throw
std
::
runtime_error
(
"No peer found for fd: "
+
to_string
(
fd
));
}
return
it
->
second
;
}
void
void
IoWorker
::
readIncomingData
(
int
fd
)
{
IoWorker
::
handleIncoming
(
const
std
::
shared_ptr
<
Peer
>&
peer
)
{
char
buffer
[
Const
::
MaxBuffer
];
char
buffer
[
Const
::
MaxBuffer
];
memset
(
buffer
,
0
,
sizeof
buffer
);
memset
(
buffer
,
0
,
sizeof
buffer
);
ssize_t
totalBytes
=
0
;
ssize_t
totalBytes
=
0
;
int
fd
=
peer
->
fd
();
for
(;;)
{
for
(;;)
{
...
@@ -56,14 +97,15 @@ IoWorker::readIncomingData(int fd) {
...
@@ -56,14 +97,15 @@ IoWorker::readIncomingData(int fd) {
bytes
=
recv
(
fd
,
buffer
+
totalBytes
,
Const
::
MaxBuffer
-
totalBytes
,
0
);
bytes
=
recv
(
fd
,
buffer
+
totalBytes
,
Const
::
MaxBuffer
-
totalBytes
,
0
);
if
(
bytes
==
-
1
)
{
if
(
bytes
==
-
1
)
{
if
(
errno
==
EAGAIN
||
errno
==
EWOULDBLOCK
)
{
if
(
errno
==
EAGAIN
||
errno
==
EWOULDBLOCK
)
{
cout
<<
"Received "
<<
buffer
<<
endl
;
handler_
->
onInput
(
buffer
,
totalBytes
,
*
peer
)
;
}
else
{
}
else
{
throw
std
::
runtime_error
(
strerror
(
errno
));
throw
std
::
runtime_error
(
strerror
(
errno
));
}
}
break
;
break
;
}
}
else
if
(
bytes
==
0
)
{
else
if
(
bytes
==
0
)
{
cout
<<
"Peer has been shutdowned"
<<
endl
;
cout
<<
"Peer "
<<
*
peer
<<
" has disconnected"
<<
endl
;
close
(
fd
);
break
;
break
;
}
}
...
@@ -78,6 +120,21 @@ IoWorker::readIncomingData(int fd) {
...
@@ -78,6 +120,21 @@ IoWorker::readIncomingData(int fd) {
}
}
void
IoWorker
::
handleNewPeer
(
const
std
::
shared_ptr
<
Peer
>&
peer
)
{
std
::
cout
<<
"New peer: "
<<
*
peer
<<
std
::
endl
;
int
fd
=
peer
->
fd
();
struct
epoll_event
event
;
event
.
events
=
EPOLLIN
;
event
.
data
.
fd
=
fd
;
epoll_ctl
(
epoll_fd
,
EPOLL_CTL_ADD
,
fd
,
&
event
);
peers
.
insert
(
std
::
make_pair
(
fd
,
peer
));
}
void
void
IoWorker
::
run
()
{
IoWorker
::
run
()
{
struct
epoll_event
events
[
Const
::
MaxEvents
];
struct
epoll_event
events
[
Const
::
MaxEvents
];
...
@@ -89,21 +146,19 @@ IoWorker::run() {
...
@@ -89,21 +146,19 @@ IoWorker::run() {
break
;
break
;
case
0
:
case
0
:
if
(
!
mailbox
.
isEmpty
())
{
if
(
!
mailbox
.
isEmpty
())
{
int
*
fd
=
mailbox
.
clear
();
std
::
unique_ptr
<
Message
>
msg
(
mailbox
.
clear
());
if
(
msg
->
type
()
==
Message
::
Type
::
NewPeer
)
{
struct
epoll_event
event
;
auto
peer_msg
=
message_cast
<
PeerMessage
>
(
msg
);
event
.
events
=
EPOLLIN
;
event
.
data
.
fd
=
*
fd
;
epoll_ctl
(
epoll_fd
,
EPOLL_CTL_ADD
,
*
fd
,
&
event
);
handleNewPeer
(
peer_msg
->
peer
());
}
delete
fd
;
}
}
break
;
break
;
default:
default:
for
(
int
i
=
0
;
i
<
ready_fds
;
++
i
)
{
for
(
int
i
=
0
;
i
<
ready_fds
;
++
i
)
{
const
struct
epoll_event
*
event
=
events
+
i
;
const
struct
epoll_event
*
event
=
events
+
i
;
readIncomingData
(
event
->
data
.
fd
);
handleIncoming
(
getPeer
(
event
->
data
.
fd
)
);
}
}
break
;
break
;
...
@@ -111,13 +166,62 @@ IoWorker::run() {
...
@@ -111,13 +166,62 @@ IoWorker::run() {
}
}
}
}
Handler
::
Handler
()
{
}
Handler
::~
Handler
()
{
}
void
Handler
::
onConnection
()
{
}
void
Handler
::
onDisconnection
()
{
}
Listener
::
Listener
()
:
listen_fd
(
-
1
)
{
}
Listener
::
Listener
(
const
Address
&
address
)
Listener
::
Listener
(
const
Address
&
address
)
:
addr_
(
address
)
:
addr_
(
address
)
,
listen_fd
(
-
1
)
,
listen_fd
(
-
1
)
{
{
}
}
bool
Listener
::
bind
()
{
void
Listener
::
init
(
size_t
workers
)
{
if
(
workers
>
hardware_concurrency
())
{
// Log::warning() << "More workers than available cores"
}
for
(
size_t
i
=
0
;
i
<
workers
;
++
i
)
{
auto
wrk
=
std
::
unique_ptr
<
IoWorker
>
(
new
IoWorker
);
ioGroup
.
push_back
(
std
::
move
(
wrk
));
}
}
void
Listener
::
setHandler
(
const
std
::
shared_ptr
<
Handler
>&
handler
)
{
handler_
=
handler
;
}
bool
Listener
::
bind
()
{
return
bind
(
addr_
);
}
bool
Listener
::
bind
(
const
Address
&
address
)
{
if
(
ioGroup
.
empty
())
{
throw
std
::
runtime_error
(
"Call init() before calling bind()"
);
}
addr_
=
address
;
struct
addrinfo
hints
;
struct
addrinfo
hints
;
hints
.
ai_family
=
AF_INET
;
hints
.
ai_family
=
AF_INET
;
...
@@ -135,7 +239,7 @@ bool Listener::bind() {
...
@@ -135,7 +239,7 @@ bool Listener::bind() {
char
port
[
MaxPortLen
];
char
port
[
MaxPortLen
];
std
::
fill
(
port
,
port
+
MaxPortLen
,
0
);
std
::
fill
(
port
,
port
+
MaxPortLen
,
0
);
std
::
snprintf
(
port
,
MaxPortLen
,
"%d"
,
addr_
.
port
(
));
std
::
snprintf
(
port
,
MaxPortLen
,
"%d"
,
static_cast
<
uint16_t
>
(
addr_
.
port
()
));
struct
addrinfo
*
addrs
;
struct
addrinfo
*
addrs
;
TRY
(
::
getaddrinfo
(
host
.
c_str
(),
port
,
&
hints
,
&
addrs
));
TRY
(
::
getaddrinfo
(
host
.
c_str
(),
port
,
&
hints
,
&
addrs
));
...
@@ -156,16 +260,15 @@ bool Listener::bind() {
...
@@ -156,16 +260,15 @@ bool Listener::bind() {
listen_fd
=
fd
;
listen_fd
=
fd
;
for
(
int
i
=
0
;
i
<
4
;
++
i
)
{
for
(
auto
&
io
:
ioGroup
)
{
auto
wrk
=
std
::
unique_ptr
<
IoWorker
>
(
new
IoWorker
);
io
->
start
(
handler_
);
wrk
->
start
();
ioGroup
.
push_back
(
std
::
move
(
wrk
));
}
}
return
true
;
return
true
;
}
}
void
Listener
::
run
()
{
void
Listener
::
run
()
{
for
(;;)
{
for
(;;)
{
struct
sockaddr_in
peer_addr
;
struct
sockaddr_in
peer_addr
;
socklen_t
peer_addr_len
=
sizeof
(
peer_addr
);
socklen_t
peer_addr_len
=
sizeof
(
peer_addr
);
...
@@ -177,31 +280,40 @@ void Listener::run() {
...
@@ -177,31 +280,40 @@ void Listener::run() {
if
(
getnameinfo
((
struct
sockaddr
*
)
&
peer_addr
,
peer_addr_len
,
peer_host
,
NI_MAXHOST
,
nullptr
,
0
,
0
)
==
0
)
{
if
(
getnameinfo
((
struct
sockaddr
*
)
&
peer_addr
,
peer_addr_len
,
peer_host
,
NI_MAXHOST
,
nullptr
,
0
,
0
)
==
0
)
{
Address
addr
=
Address
::
fromUnix
((
struct
sockaddr
*
)
&
peer_addr
);
Address
addr
=
Address
::
fromUnix
((
struct
sockaddr
*
)
&
peer_addr
);
Peer
peer
(
addr
,
peer_host
);
auto
peer
=
make_shared
<
Peer
>
(
addr
,
peer_host
);
peer
->
associateFd
(
client_fd
);
std
::
cout
<<
"New peer: "
<<
peer
<<
std
::
endl
;
dispatchPeer
(
peer
);
dispatchConnection
(
client_fd
);
}
}
}
}
}
}
Address
Listener
::
address
()
const
{
return
addr_
;
}
void
Listener
::
dispatchPeer
(
const
std
::
shared_ptr
<
Peer
>&
peer
)
{
const
size_t
workers
=
ioGroup
.
size
();
size_t
start
=
peer
->
fd
()
%
workers
;
void
Listener
::
dispatchConnection
(
int
fd
)
{
size_t
start
=
fd
%
4
;
/* Find the first available worker */
/* Find the first available worker */
size_t
current
=
start
;
size_t
current
=
start
;
for
(;;)
{
for
(;;)
{
auto
&
mailbox
=
ioGroup
[
current
]
->
mailbox
;
auto
&
mailbox
=
ioGroup
[
current
]
->
mailbox
;
if
(
mailbox
.
isEmpty
())
{
if
(
mailbox
.
isEmpty
())
{
int
*
message
=
new
int
(
fd
);
auto
message
=
new
PeerMessage
(
peer
);
int
*
old
=
mailbox
.
post
(
message
);
auto
*
old
=
mailbox
.
post
(
message
);
assert
(
old
==
nullptr
);
assert
(
old
==
nullptr
);
return
;
return
;
}
}
current
=
(
current
+
1
)
%
4
;
current
=
(
current
+
1
)
%
workers
;
if
(
current
==
start
)
{
if
(
current
==
start
)
{
break
;
break
;
}
}
...
...
src/listener.h
View file @
9fcc1835
...
@@ -8,45 +8,81 @@
...
@@ -8,45 +8,81 @@
#include "net.h"
#include "net.h"
#include "mailbox.h"
#include "mailbox.h"
#include "os.h"
#include <vector>
#include <vector>
#include <memory>
#include <memory>
#include <thread>
#include <thread>
#include <unordered_map>
namespace
Net
{
namespace
Net
{
namespace
Tcp
{
namespace
Tcp
{
class
Peer
;
class
Message
;
class
Handler
;
class
IoWorker
{
class
IoWorker
{
public:
public:
Mailbox
<
int
>
mailbox
;
Mailbox
<
Message
>
mailbox
;
IoWorker
();
IoWorker
();
~
IoWorker
();
~
IoWorker
();
void
start
();
void
start
(
const
std
::
shared_ptr
<
Handler
>
&
handler
);
private:
private:
int
epoll_fd
;
int
epoll_fd
;
std
::
unique_ptr
<
std
::
thread
>
thread
;
std
::
unique_ptr
<
std
::
thread
>
thread
;
std
::
unordered_map
<
Fd
,
std
::
shared_ptr
<
Peer
>>
peers
;
std
::
shared_ptr
<
Handler
>
handler_
;
std
::
shared_ptr
<
Peer
>
getPeer
(
Fd
fd
)
const
;
void
handleIncoming
(
const
std
::
shared_ptr
<
Peer
>&
peer
);
void
handleNewPeer
(
const
std
::
shared_ptr
<
Peer
>&
peer
);
void
run
();
};
class
Listener
{
public:
friend
class
IoWorker
;
Listener
();
Listener
(
const
Address
&
address
);
void
init
(
size_t
workers
);
void
setHandler
(
const
std
::
shared_ptr
<
Handler
>&
handler
);
bool
bind
();
bool
bind
(
const
Address
&
adress
);
void
readIncomingData
(
int
fd
);
void
run
();
void
run
();
Address
address
()
const
;
private:
Address
addr_
;
int
listen_fd
;
std
::
vector
<
std
::
unique_ptr
<
IoWorker
>>
ioGroup
;
std
::
shared_ptr
<
Handler
>
handler_
;
void
dispatchPeer
(
const
std
::
shared_ptr
<
Peer
>&
peer
);
};
};
class
Listener
{
class
Handler
{
public:
public:
Listener
(
const
Address
&
address
);
Handler
();
~
Handler
();
bool
bind
()
;
virtual
void
onInput
(
const
char
*
buffer
,
size_t
len
,
Tcp
::
Peer
&
peer
)
=
0
;
void
run
()
;
virtual
void
onOutput
()
=
0
;
private:
virtual
void
onConnection
();
virtual
void
onDisconnection
();
Address
addr_
;
};
int
listen_fd
;
std
::
vector
<
std
::
unique_ptr
<
IoWorker
>>
ioGroup
;
void
dispatchConnection
(
int
fd
);
};
}
// namespace Tcp
}
// namespace Tcp
...
...
src/mailbox.h
View file @
9fcc1835
...
@@ -13,6 +13,10 @@
...
@@ -13,6 +13,10 @@
template
<
typename
T
>
template
<
typename
T
>
class
Mailbox
{
class
Mailbox
{
public:
public:
Mailbox
()
{
data
.
store
(
nullptr
);
}
const
T
*
get
()
const
{
const
T
*
get
()
const
{
if
(
isEmpty
())
{
if
(
isEmpty
())
{
throw
std
::
runtime_error
(
"Can not retrieve mail from empty mailbox"
);
throw
std
::
runtime_error
(
"Can not retrieve mail from empty mailbox"
);
...
...
src/net.cc
View file @
9fcc1835
...
@@ -9,7 +9,6 @@
...
@@ -9,7 +9,6 @@
#include <stdexcept>
#include <stdexcept>
#include <limits>
#include <limits>
#include <cstring>
#include <cstring>
#include <fcntl.h>
#include <netinet/in.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <arpa/inet.h>
...
@@ -17,16 +16,19 @@ using namespace std;
...
@@ -17,16 +16,19 @@ using namespace std;
namespace
Net
{
namespace
Net
{
bool
make_non_blocking
(
int
sfd
)
Port
::
Port
(
uint16_t
port
)
{
:
port
(
port
)
int
flags
=
fcntl
(
sfd
,
F_GETFL
,
0
);
{
}
if
(
flags
==
-
1
)
return
false
;
flags
|=
O_NONBLOCK
;
bool
int
ret
=
fcntl
(
sfd
,
F_SETFL
,
flags
);
Port
::
isReserved
()
const
{
if
(
ret
==
-
1
)
return
false
;
return
port
<
1024
;
}
return
true
;
bool
Port
::
isUsed
()
const
{
throw
std
::
runtime_error
(
"Unimplemented"
);
return
false
;
}
}
Ipv4
::
Ipv4
(
uint8_t
a
,
uint8_t
b
,
uint8_t
c
,
uint8_t
d
)
Ipv4
::
Ipv4
(
uint8_t
a
,
uint8_t
b
,
uint8_t
c
,
uint8_t
d
)
...
@@ -36,11 +38,13 @@ Ipv4::Ipv4(uint8_t a, uint8_t b, uint8_t c, uint8_t d)
...
@@ -36,11 +38,13 @@ Ipv4::Ipv4(uint8_t a, uint8_t b, uint8_t c, uint8_t d)
,
d
(
d
)
,
d
(
d
)
{
}
{
}
Ipv4
Ipv4
::
any
()
{
Ipv4
Ipv4
::
any
()
{
return
Ipv4
(
0
,
0
,
0
,
0
);
return
Ipv4
(
0
,
0
,
0
,
0
);
}
}
std
::
string
Ipv4
::
toString
()
const
{
std
::
string
Ipv4
::
toString
()
const
{
static
constexpr
size_t
MaxSize
=
sizeof
(
"255"
)
*
4
+
3
+
1
;
/* 4 * 255 + 3 * dot + \0 */
static
constexpr
size_t
MaxSize
=
sizeof
(
"255"
)
*
4
+
3
+
1
;
/* 4 * 255 + 3 * dot + \0 */
char
buff
[
MaxSize
];
char
buff
[
MaxSize
];
...
@@ -88,6 +92,7 @@ Address
...
@@ -88,6 +92,7 @@ Address
Address
::
fromUnix
(
struct
sockaddr
*
addr
)
{
Address
::
fromUnix
(
struct
sockaddr
*
addr
)
{
struct
sockaddr_in
*
in_addr
=
reinterpret_cast
<
struct
sockaddr_in
*>
(
addr
);
struct
sockaddr_in
*
in_addr
=
reinterpret_cast
<
struct
sockaddr_in
*>
(
addr
);
std
::
string
host
=
TRY_RET
(
inet_ntoa
(
in_addr
->
sin_addr
));
std
::
string
host
=
TRY_RET
(
inet_ntoa
(
in_addr
->
sin_addr
));
int
port
=
in_addr
->
sin_port
;
int
port
=
in_addr
->
sin_port
;
return
Address
(
std
::
move
(
host
),
port
);
return
Address
(
std
::
move
(
host
),
port
);
...
...
src/net.h
View file @
9fcc1835
...
@@ -10,9 +10,18 @@
...
@@ -10,9 +10,18 @@
namespace
Net
{
namespace
Net
{
typedef
uint16_t
Port
;
class
Port
{
public:
Port
(
uint16_t
port
=
0
);
operator
uint16_t
()
const
{
return
port
;
}
bool
isReserved
()
const
;
bool
isUsed
()
const
;
bool
make_non_blocking
(
int
fd
);
private:
uint16_t
port
;
};
class
Ipv4
{
class
Ipv4
{
public:
public:
...
...
src/peer.cc
View file @
9fcc1835
...
@@ -5,17 +5,22 @@
...
@@ -5,17 +5,22 @@
#include "peer.h"
#include "peer.h"
#include <iostream>
#include <iostream>
#include <stdexcept>
namespace
Net
{
namespace
Net
{
namespace
Tcp
{
using
namespace
std
;
using
namespace
std
;
Peer
::
Peer
()
Peer
::
Peer
()
:
fd_
(
-
1
)
{
}
{
}
Peer
::
Peer
(
const
Address
&
addr
,
const
string
&
host
)
Peer
::
Peer
(
const
Address
&
addr
,
const
string
&
host
)
:
addr
(
addr
)
:
addr
(
addr
)
,
host
(
host
)
,
host
(
host
)
,
fd_
(
-
1
)
{
}
{
}
Address
Address
...
@@ -28,10 +33,26 @@ Peer::hostname() const {
...
@@ -28,10 +33,26 @@ Peer::hostname() const {
return
host
;
return
host
;
}
}
void
Peer
::
associateFd
(
int
fd
)
{
fd_
=
fd
;
}
int
Peer
::
fd
()
const
{
if
(
fd_
==
-
1
)
{
throw
std
::
runtime_error
(
"The peer has no associated fd"
);
}
return
fd_
;
}
std
::
ostream
&
operator
<<
(
std
::
ostream
&
os
,
const
Peer
&
peer
)
{
std
::
ostream
&
operator
<<
(
std
::
ostream
&
os
,
const
Peer
&
peer
)
{
const
auto
&
addr
=
peer
.
address
();
const
auto
&
addr
=
peer
.
address
();
os
<<
"("
<<
addr
.
host
()
<<
", "
<<
addr
.
port
()
<<
") ["
<<
peer
.
hostname
()
<<
"]"
;
os
<<
"("
<<
addr
.
host
()
<<
", "
<<
addr
.
port
()
<<
") ["
<<
peer
.
hostname
()
<<
"]"
;
return
os
;
return
os
;
}
}
}
// namespace Tcp
}
// namespace Net
}
// namespace Net
src/peer.h
View file @
9fcc1835
...
@@ -7,25 +7,33 @@
...
@@ -7,25 +7,33 @@
#pragma once
#pragma once
#include "net.h"
#include "net.h"
#include "os.h"
#include <string>
#include <string>
#include <iostream>
#include <iostream>
namespace
Net
{
namespace
Net
{
namespace
Tcp
{
class
Peer
{
class
Peer
{
public:
public:
Peer
();
Peer
();
Peer
(
const
Address
&
addr
,
const
std
::
string
&
hostname
);
Peer
(
const
Address
&
addr
,
const
std
::
string
&
hostname
);
Address
address
()
const
;
Address
address
()
const
;
std
::
string
hostname
()
const
;
std
::
string
hostname
()
const
;
void
associateFd
(
Fd
fd
);
Fd
fd
()
const
;
private:
private:
Address
addr
;
Address
addr
;
std
::
string
host
;
std
::
string
host
;
Fd
fd_
;
};
};
std
::
ostream
&
operator
<<
(
std
::
ostream
&
os
,
const
Peer
&
peer
);
std
::
ostream
&
operator
<<
(
std
::
ostream
&
os
,
const
Peer
&
peer
);
}
// namespace Tcp
}
// namespace Net
}
// namespace Net
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment