Skip to content
GitLab
Menu
Projects
Groups
Snippets
Loading...
Help
Help
Support
Community forum
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in / Register
Toggle navigation
Menu
Open sidebar
ruanhaishen
redis
Commits
72f30bcd
Commit
72f30bcd
authored
Mar 26, 2012
by
Dave Pacheco
Committed by
Pieter Noordhuis
May 14, 2012
Browse files
use port_getn instead of port_get
parent
2aa1efb8
Changes
1
Hide whitespace changes
Inline
Side-by-side
src/ae_evport.c
View file @
72f30bcd
...
@@ -12,7 +12,7 @@
...
@@ -12,7 +12,7 @@
#include <stdio.h>
#include <stdio.h>
int
evport_debug
=
0
;
static
int
evport_debug
=
0
;
/*
/*
* This file implements the ae API using event ports, present on Solaris-based
* This file implements the ae API using event ports, present on Solaris-based
...
@@ -21,33 +21,33 @@ int evport_debug = 0;
...
@@ -21,33 +21,33 @@ int evport_debug = 0;
* events that the consumer is interested in (e.g., POLLIN and POLLOUT).
* events that the consumer is interested in (e.g., POLLIN and POLLOUT).
*
*
* There's one tricky piece to this implementation: when we return events via
* There's one tricky piece to this implementation: when we return events via
* aeApiPoll, the corresponding file descriptor become
s
dissociated from the
* aeApiPoll, the corresponding file descriptor
s
become dissociated from the
* port. This is necessary because poll events are level-triggered, so if the
* port. This is necessary because poll events are level-triggered, so if the
* fd didn't become dissociated, it would immediately fire another event since
* fd didn't become dissociated, it would immediately fire another event since
* the underlying state hasn't changed yet. We must reassociate the file
* the underlying state hasn't changed yet. We must reassociate the file
* descriptor, but only after we know that our caller has actually read from it.
* descriptor, but only after we know that our caller has actually read from it.
* The ae API does not tell us exactly when that happens, but we do know that
* The ae API does not tell us exactly when that happens, but we do know that
* it must happen by the time aeApiPoll is called again. Our solution is to
* it must happen by the time aeApiPoll is called again. Our solution is to
* keep track of the last fd returned by aeApiPoll and reassociate
i
t next
time
* keep track of the last fd
s
returned by aeApiPoll and reassociate t
hem
next
* aeApiPoll is invoked.
*
time
aeApiPoll is invoked.
*
*
* To summarize, in this module, each fd association is EITHER (a) represented
* To summarize, in this module, each fd association is EITHER (a) represented
* only via the in-kernel assocation OR (b) represented by pending_fd and
* only via the in-kernel assocation OR (b) represented by pending_fd
s
and
* pending_mask. (b) is only true for the last fd we returned from aeApiPoll,
* pending_mask
s
. (b) is only true for the last fd
s
we returned from aeApiPoll,
* and only until we enter aeApiPoll again (at which point we restore the
* and only until we enter aeApiPoll again (at which point we restore the
* in-kernel association).
* in-kernel association).
*
* We currently only return one fd event per call to aeApiPoll. This could be
* extended to return more than one by extending the corresponding pending
* fields and using port_getn().
*/
*/
#define MAX_EVENT_BATCHSZ 512
typedef
struct
aeApiState
{
typedef
struct
aeApiState
{
int
portfd
;
/* event port */
int
portfd
;
/* event port */
int
pending_fd
;
/* pending fd */
int
npending
;
/* # of pending fds */
int
pending_mask
;
/* pending fd's mask */
int
pending_fds
[
MAX_EVENT_BATCHSZ
];
/* pending fds */
int
pending_masks
[
MAX_EVENT_BATCHSZ
];
/* pending fds' masks */
}
aeApiState
;
}
aeApiState
;
static
int
aeApiCreate
(
aeEventLoop
*
eventLoop
)
{
static
int
aeApiCreate
(
aeEventLoop
*
eventLoop
)
{
int
i
;
aeApiState
*
state
=
zmalloc
(
sizeof
(
aeApiState
));
aeApiState
*
state
=
zmalloc
(
sizeof
(
aeApiState
));
if
(
!
state
)
return
-
1
;
if
(
!
state
)
return
-
1
;
...
@@ -57,8 +57,12 @@ static int aeApiCreate(aeEventLoop *eventLoop) {
...
@@ -57,8 +57,12 @@ static int aeApiCreate(aeEventLoop *eventLoop) {
return
-
1
;
return
-
1
;
}
}
state
->
pending_fd
=
-
1
;
state
->
npending
=
0
;
state
->
pending_mask
=
AE_NONE
;
for
(
i
=
0
;
i
<
MAX_EVENT_BATCHSZ
;
i
++
)
{
state
->
pending_fds
[
i
]
=
-
1
;
state
->
pending_masks
[
i
]
=
AE_NONE
;
}
eventLoop
->
apidata
=
state
;
eventLoop
->
apidata
=
state
;
return
0
;
return
0
;
...
@@ -71,11 +75,21 @@ static void aeApiFree(aeEventLoop *eventLoop) {
...
@@ -71,11 +75,21 @@ static void aeApiFree(aeEventLoop *eventLoop) {
zfree
(
state
);
zfree
(
state
);
}
}
static
int
aeApiLookupPending
(
aeApiState
*
state
,
int
fd
)
{
int
i
;
for
(
i
=
0
;
i
<
state
->
npending
;
i
++
)
{
if
(
state
->
pending_fds
[
i
]
==
fd
)
return
(
i
);
}
return
(
-
1
);
}
/*
/*
* Helper function to invoke port_associate for the given fd and mask.
* Helper function to invoke port_associate for the given fd and mask.
*/
*/
static
int
aeApiAssociate
(
const
char
*
where
,
int
portfd
,
int
fd
,
int
mask
)
static
int
aeApiAssociate
(
const
char
*
where
,
int
portfd
,
int
fd
,
int
mask
)
{
{
int
events
=
0
;
int
events
=
0
;
int
rv
,
err
;
int
rv
,
err
;
...
@@ -87,7 +101,8 @@ static int aeApiAssociate(const char *where, int portfd, int fd, int mask)
...
@@ -87,7 +101,8 @@ static int aeApiAssociate(const char *where, int portfd, int fd, int mask)
if
(
evport_debug
)
if
(
evport_debug
)
fprintf
(
stderr
,
"%s: port_associate(%d, 0x%x) = "
,
where
,
fd
,
events
);
fprintf
(
stderr
,
"%s: port_associate(%d, 0x%x) = "
,
where
,
fd
,
events
);
rv
=
port_associate
(
portfd
,
PORT_SOURCE_FD
,
fd
,
events
,
(
void
*
)
mask
);
rv
=
port_associate
(
portfd
,
PORT_SOURCE_FD
,
fd
,
events
,
(
void
*
)(
uintptr_t
)
mask
);
err
=
errno
;
err
=
errno
;
if
(
evport_debug
)
if
(
evport_debug
)
...
@@ -105,7 +120,7 @@ static int aeApiAssociate(const char *where, int portfd, int fd, int mask)
...
@@ -105,7 +120,7 @@ static int aeApiAssociate(const char *where, int portfd, int fd, int mask)
static
int
aeApiAddEvent
(
aeEventLoop
*
eventLoop
,
int
fd
,
int
mask
)
{
static
int
aeApiAddEvent
(
aeEventLoop
*
eventLoop
,
int
fd
,
int
mask
)
{
aeApiState
*
state
=
eventLoop
->
apidata
;
aeApiState
*
state
=
eventLoop
->
apidata
;
int
fullmask
;
int
fullmask
,
pfd
;
if
(
evport_debug
)
if
(
evport_debug
)
fprintf
(
stderr
,
"aeApiAddEvent: fd %d mask 0x%x
\n
"
,
fd
,
mask
);
fprintf
(
stderr
,
"aeApiAddEvent: fd %d mask 0x%x
\n
"
,
fd
,
mask
);
...
@@ -116,8 +131,9 @@ static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
...
@@ -116,8 +131,9 @@ static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
* we call port_associate() again.
* we call port_associate() again.
*/
*/
fullmask
=
mask
|
eventLoop
->
events
[
fd
].
mask
;
fullmask
=
mask
|
eventLoop
->
events
[
fd
].
mask
;
pfd
=
aeApiLookupPending
(
state
,
fd
);
if
(
fd
=
=
state
->
pending_fd
)
{
if
(
p
fd
!
=
-
1
)
{
/*
/*
* This fd was recently returned from aeApiPoll. It should be safe to
* This fd was recently returned from aeApiPoll. It should be safe to
* assume that the consumer has processed that poll event, but we play
* assume that the consumer has processed that poll event, but we play
...
@@ -126,7 +142,7 @@ static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
...
@@ -126,7 +142,7 @@ static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
*/
*/
if
(
evport_debug
)
if
(
evport_debug
)
fprintf
(
stderr
,
"aeApiAddEvent: adding to pending fd %d
\n
"
,
fd
);
fprintf
(
stderr
,
"aeApiAddEvent: adding to pending fd %d
\n
"
,
fd
);
state
->
pending_mask
|=
fullmask
;
state
->
pending_mask
s
[
pfd
]
|=
fullmask
;
return
0
;
return
0
;
}
}
...
@@ -135,12 +151,14 @@ static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
...
@@ -135,12 +151,14 @@ static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
static
void
aeApiDelEvent
(
aeEventLoop
*
eventLoop
,
int
fd
,
int
mask
)
{
static
void
aeApiDelEvent
(
aeEventLoop
*
eventLoop
,
int
fd
,
int
mask
)
{
aeApiState
*
state
=
eventLoop
->
apidata
;
aeApiState
*
state
=
eventLoop
->
apidata
;
int
fullmask
;
int
fullmask
,
pfd
;
if
(
evport_debug
)
if
(
evport_debug
)
fprintf
(
stderr
,
"del fd %d mask 0x%x
\n
"
,
fd
,
mask
);
fprintf
(
stderr
,
"del fd %d mask 0x%x
\n
"
,
fd
,
mask
);
if
(
fd
==
state
->
pending_fd
)
{
pfd
=
aeApiLookupPending
(
state
,
fd
);
if
(
pfd
!=
-
1
)
{
if
(
evport_debug
)
if
(
evport_debug
)
fprintf
(
stderr
,
"deleting event from pending fd %d
\n
"
,
fd
);
fprintf
(
stderr
,
"deleting event from pending fd %d
\n
"
,
fd
);
...
@@ -149,10 +167,10 @@ static void aeApiDelEvent(aeEventLoop *eventLoop, int fd, int mask) {
...
@@ -149,10 +167,10 @@ static void aeApiDelEvent(aeEventLoop *eventLoop, int fd, int mask) {
* associated with the port. All we need to do is update
* associated with the port. All we need to do is update
* pending_mask appropriately.
* pending_mask appropriately.
*/
*/
state
->
pending_mask
&=
~
mask
;
state
->
pending_mask
s
[
pfd
]
&=
~
mask
;
if
(
state
->
pending_mask
==
AE_NONE
)
if
(
state
->
pending_mask
s
[
pfd
]
==
AE_NONE
)
state
->
pending_fd
=
-
1
;
state
->
pending_fd
s
[
pfd
]
=
-
1
;
return
;
return
;
}
}
...
@@ -194,25 +212,32 @@ static void aeApiDelEvent(aeEventLoop *eventLoop, int fd, int mask) {
...
@@ -194,25 +212,32 @@ static void aeApiDelEvent(aeEventLoop *eventLoop, int fd, int mask) {
static
int
aeApiPoll
(
aeEventLoop
*
eventLoop
,
struct
timeval
*
tvp
)
{
static
int
aeApiPoll
(
aeEventLoop
*
eventLoop
,
struct
timeval
*
tvp
)
{
aeApiState
*
state
=
eventLoop
->
apidata
;
aeApiState
*
state
=
eventLoop
->
apidata
;
struct
timespec
timeout
,
*
tsp
;
struct
timespec
timeout
,
*
tsp
;
int
mask
;
int
mask
,
i
;
port_event_t
event
;
uint_t
nevents
;
port_event_t
event
[
MAX_EVENT_BATCHSZ
];
/*
/*
* If we've returned
an
fd event before, we must reassociate
d
th
at fd
with
* If we've returned fd event
s
before, we must reassociate th
em
with
the
*
the
port now, before calling port_get(). See the block comment at the
* port now, before calling port_get(). See the block comment at the
top of
*
top of
this file for an explanation of why.
* this file for an explanation of why.
*/
*/
if
(
state
->
pending_mask
!=
AE_NONE
)
{
for
(
i
=
0
;
i
<
state
->
npending
;
i
++
)
{
if
(
aeApiAssociate
(
"aeApiPoll"
,
state
->
portfd
,
state
->
pending_fd
,
if
(
state
->
pending_fds
[
i
]
==
-
1
)
state
->
pending_mask
)
!=
0
)
{
/* This fd has since been deleted. */
continue
;
if
(
aeApiAssociate
(
"aeApiPoll"
,
state
->
portfd
,
state
->
pending_fds
[
i
],
state
->
pending_masks
[
i
])
!=
0
)
{
/* See aeApiDelEvent for why this case is fatal. */
/* See aeApiDelEvent for why this case is fatal. */
abort
();
abort
();
}
}
state
->
pending_mask
=
AE_NONE
;
state
->
pending_mask
s
[
i
]
=
AE_NONE
;
state
->
pending_fd
=
-
1
;
state
->
pending_fd
s
[
i
]
=
-
1
;
}
}
state
->
npending
=
0
;
if
(
tvp
!=
NULL
)
{
if
(
tvp
!=
NULL
)
{
timeout
.
tv_sec
=
tvp
->
tv_sec
;
timeout
.
tv_sec
=
tvp
->
tv_sec
;
timeout
.
tv_nsec
=
tvp
->
tv_usec
*
1000
;
timeout
.
tv_nsec
=
tvp
->
tv_usec
*
1000
;
...
@@ -221,7 +246,13 @@ static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
...
@@ -221,7 +246,13 @@ static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
tsp
=
NULL
;
tsp
=
NULL
;
}
}
if
(
port_get
(
state
->
portfd
,
&
event
,
tsp
)
==
-
1
)
{
/*
* port_getn can return with errno == ETIME having returned some events (!).
* So if we get ETIME, we check nevents, too.
*/
nevents
=
1
;
if
(
port_getn
(
state
->
portfd
,
event
,
MAX_EVENT_BATCHSZ
,
&
nevents
,
tsp
)
==
-
1
&&
(
errno
!=
ETIME
||
nevents
==
0
))
{
if
(
errno
==
ETIME
||
errno
==
EINTR
)
if
(
errno
==
ETIME
||
errno
==
EINTR
)
return
0
;
return
0
;
...
@@ -230,23 +261,27 @@ static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
...
@@ -230,23 +261,27 @@ static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
abort
();
abort
();
}
}
mask
=
0
;
state
->
npending
=
nevents
;
if
(
event
.
portev_events
&
POLLIN
)
mask
|=
AE_READABLE
;
if
(
event
.
portev_events
&
POLLOUT
)
mask
|=
AE_WRITABLE
;
eventLoop
->
fired
[
0
].
fd
=
event
.
portev_object
;
for
(
i
=
0
;
i
<
nevents
;
i
++
)
{
eventLoop
->
fired
[
0
].
mask
=
mask
;
mask
=
0
;
if
(
event
[
i
].
portev_events
&
POLLIN
)
mask
|=
AE_READABLE
;
if
(
event
[
i
].
portev_events
&
POLLOUT
)
mask
|=
AE_WRITABLE
;
if
(
evport_debug
)
eventLoop
->
fired
[
i
].
fd
=
event
[
i
].
portev_object
;
fprintf
(
stderr
,
"aeApiPoll: fd %d mask 0x%x
\n
"
,
event
.
portev_object
,
eventLoop
->
fired
[
i
].
mask
=
mask
;
mask
);
state
->
pending_fd
=
event
.
portev_object
;
if
(
evport_debug
)
state
->
pending_mask
=
(
uintptr_t
)
event
.
portev_user
;
fprintf
(
stderr
,
"aeApiPoll: fd %d mask 0x%x
\n
"
,
(
int
)
event
[
i
].
portev_object
,
mask
);
state
->
pending_fds
[
i
]
=
event
[
i
].
portev_object
;
state
->
pending_masks
[
i
]
=
(
uintptr_t
)
event
[
i
].
portev_user
;
}
return
1
;
return
nevents
;
}
}
static
char
*
aeApiName
(
void
)
{
static
char
*
aeApiName
(
void
)
{
...
...
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