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
145f472c
Commit
145f472c
authored
Dec 09, 2015
by
Mathieu Stefani
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Started working on a way to get I/O threads load
parent
4684f678
Changes
6
Hide whitespace changes
Inline
Side-by-side
Showing
6 changed files
with
155 additions
and
1 deletion
+155
-1
src/io.cc
src/io.cc
+25
-1
src/io.h
src/io.h
+24
-0
src/listener.cc
src/listener.cc
+47
-0
src/listener.h
src/listener.h
+8
-0
src/os.cc
src/os.cc
+35
-0
src/os.h
src/os.h
+16
-0
No files found.
src/io.cc
View file @
145f472c
...
...
@@ -231,6 +231,9 @@ IoWorker::run() {
mailbox
.
bind
(
poller
);
auto
n
=
notifier
.
tag
();
poller
.
addFd
(
n
.
value
(),
NotifyOn
::
Read
,
n
,
Polling
::
Mode
::
Edge
);
std
::
chrono
::
milliseconds
timeout
(
-
1
);
for
(;;)
{
...
...
@@ -250,7 +253,11 @@ IoWorker::run() {
if
(
msg
->
type
()
==
Message
::
Type
::
Shutdown
)
{
return
;
}
}
else
{
}
else
if
(
event
.
tag
==
notifier
.
tag
())
{
handleNotify
();
}
else
{
if
(
event
.
flags
.
hasFlag
(
NotifyOn
::
Read
))
{
auto
fd
=
event
.
tag
.
value
();
if
(
fd
==
timerFd
)
{
...
...
@@ -314,6 +321,23 @@ IoWorker::handleTimeout() {
});
}
void
IoWorker
::
handleNotify
()
{
optionally_do
(
load
,
[
&
](
const
Load
&
entry
)
{
while
(
this
->
notifier
.
tryRead
())
;
rusage
now
;
auto
res
=
getrusage
(
RUSAGE_THREAD
,
&
now
);
if
(
res
==
-
1
)
entry
.
reject
(
std
::
runtime_error
(
"Could not compute usage"
));
entry
.
resolve
(
now
);
});
load
=
None
();
}
Async
::
Promise
<
ssize_t
>
IoWorker
::
asyncWrite
(
Fd
fd
,
const
void
*
buf
,
size_t
len
)
{
return
Async
::
Promise
<
ssize_t
>
([
=
](
Async
::
Resolver
&
resolve
,
Async
::
Rejection
&
reject
)
{
...
...
src/io.h
View file @
145f472c
...
...
@@ -16,6 +16,8 @@
#include <mutex>
#include <memory>
#include <unordered_map>
#include <sys/time.h>
#include <sys/resource.h>
namespace
Net
{
...
...
@@ -49,6 +51,13 @@ public:
void
disarmTimer
();
Async
::
Promise
<
rusage
>
getLoad
()
{
return
Async
::
Promise
<
rusage
>
([
=
](
Async
::
Resolver
&
resolve
,
Async
::
Rejection
&
reject
)
{
this
->
load
=
Some
(
Load
(
std
::
move
(
resolve
),
std
::
move
(
reject
)));
this
->
notifier
.
notify
();
});
}
private:
struct
OnHoldWrite
{
OnHoldWrite
(
Async
::
Resolver
resolve
,
Async
::
Rejection
reject
,
...
...
@@ -84,6 +93,17 @@ private:
Async
::
Rejection
reject
;
};
struct
Load
{
Load
(
Async
::
Resolver
resolve
,
Async
::
Rejection
reject
)
:
resolve
(
std
::
move
(
resolve
))
,
reject
(
std
::
move
(
reject
))
{
}
Async
::
Resolver
resolve
;
Async
::
Rejection
reject
;
};
Polling
::
Epoll
poller
;
std
::
unique_ptr
<
std
::
thread
>
thread
;
mutable
std
::
mutex
peersMutex
;
...
...
@@ -91,8 +111,11 @@ private:
std
::
unordered_map
<
Fd
,
OnHoldWrite
>
toWrite
;
Optional
<
Timer
>
timer
;
Optional
<
Load
>
load
;
Fd
timerFd
;
NotifyFd
notifier
;
std
::
shared_ptr
<
Handler
>
handler_
;
Flags
<
Options
>
options_
;
...
...
@@ -107,6 +130,7 @@ private:
void
handleIncoming
(
const
std
::
shared_ptr
<
Peer
>&
peer
);
void
handleTimeout
();
void
handleNotify
();
void
run
();
};
...
...
src/listener.cc
View file @
145f472c
...
...
@@ -175,6 +175,11 @@ Listener::bind(const Address& address) {
for
(
auto
&
io
:
ioGroup
)
{
io
->
start
(
handler_
,
options_
);
}
sh
=
false
;
loadThread
.
reset
(
new
std
::
thread
([
=
]()
{
this
->
runLoadThread
();
}));
return
true
;
}
...
...
@@ -205,11 +210,53 @@ Listener::run() {
}
}
void
Listener
::
runLoadThread
()
{
std
::
vector
<
rusage
>
lastUsages
;
while
(
!
sh
)
{
std
::
vector
<
Async
::
Promise
<
rusage
>>
loads
;
loads
.
reserve
(
ioGroup
.
size
());
for
(
const
auto
&
io
:
ioGroup
)
{
loads
.
push_back
(
io
->
getLoad
());
}
Async
::
whenAll
(
std
::
begin
(
loads
),
std
::
end
(
loads
))
.
then
([
&
](
const
std
::
vector
<
rusage
>&
usages
)
{
auto
totalElapsed
=
[](
rusage
usage
)
{
return
(
usage
.
ru_stime
.
tv_sec
*
1e6
+
usage
.
ru_stime
.
tv_usec
)
+
(
usage
.
ru_utime
.
tv_sec
*
1e6
+
usage
.
ru_utime
.
tv_usec
);
};
if
(
lastUsages
.
empty
())
lastUsages
=
usages
;
else
{
for
(
size_t
i
=
0
;
i
<
usages
.
size
();
++
i
)
{
auto
last
=
lastUsages
[
i
];
const
auto
&
usage
=
usages
[
i
];
auto
now
=
totalElapsed
(
usage
);
auto
time
=
now
-
totalElapsed
(
last
);
auto
load
=
(
time
*
100.0
)
/
1e6
;
//printf("Total load for I/O thread %lu = %.3lf%%\n", i, load);
}
lastUsages
=
usages
;
}
},
Async
::
NoExcept
);
std
::
this_thread
::
sleep_for
(
std
::
chrono
::
seconds
(
1
));
}
}
void
Listener
::
shutdown
()
{
for
(
auto
&
worker
:
ioGroup
)
{
worker
->
shutdown
();
}
sh
=
true
;
}
Address
...
...
src/listener.h
View file @
145f472c
...
...
@@ -27,6 +27,9 @@ class Listener {
public:
Listener
();
~
Listener
()
{
loadThread
->
join
();
}
Listener
(
const
Address
&
address
);
void
init
(
...
...
@@ -53,7 +56,12 @@ private:
Flags
<
Options
>
options_
;
std
::
shared_ptr
<
Handler
>
handler_
;
std
::
atomic
<
bool
>
sh
;
std
::
unique_ptr
<
std
::
thread
>
loadThread
;
void
dispatchPeer
(
const
std
::
shared_ptr
<
Peer
>&
peer
);
void
runLoadThread
();
};
}
// namespace Tcp
...
...
src/os.cc
View file @
145f472c
...
...
@@ -11,6 +11,7 @@
#include <iterator>
#include <algorithm>
#include <sys/epoll.h>
#include <sys/eventfd.h>
using
namespace
std
;
...
...
@@ -236,3 +237,37 @@ namespace Polling {
}
}
// namespace Poller
NotifyFd
::
NotifyFd
()
{
event_fd
=
TRY_RET
(
eventfd
(
0
,
EFD_NONBLOCK
|
EFD_CLOEXEC
));
}
Polling
::
Tag
NotifyFd
::
tag
()
const
{
return
Polling
::
Tag
(
event_fd
);
}
void
NotifyFd
::
notify
()
const
{
eventfd_t
val
=
1
;
TRY
(
eventfd_write
(
event_fd
,
val
));
}
void
NotifyFd
::
read
()
const
{
eventfd_t
val
;
TRY
(
eventfd_read
(
event_fd
,
&
val
));
}
bool
NotifyFd
::
tryRead
()
const
{
eventfd_t
val
;
int
res
=
eventfd_read
(
event_fd
,
&
val
);
if
(
res
==
-
1
)
{
if
(
errno
==
EAGAIN
||
errno
==
EWOULDBLOCK
)
return
false
;
throw
std
::
runtime_error
(
"Failed to read eventfd"
);
}
return
true
;
}
src/os.h
View file @
145f472c
...
...
@@ -112,3 +112,19 @@ private:
};
}
// namespace Polling
class
NotifyFd
{
public:
NotifyFd
();
Polling
::
Tag
tag
()
const
;
void
notify
()
const
;
void
read
()
const
;
bool
tryRead
()
const
;
private:
int
event_fd
;
};
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