Skip to content
Snippets Groups Projects
Commit 0e56d973 authored by Dmitry E. Oboukhov's avatar Dmitry E. Oboukhov
Browse files

try to fix race condition

parent 4d96d8e2
No related branches found
No related tags found
No related merge requests found
...@@ -35,6 +35,8 @@ const ev_tstamp IPC_TIMEOUT_INFINITY = 365*86400*100.0; ...@@ -35,6 +35,8 @@ const ev_tstamp IPC_TIMEOUT_INFINITY = 365*86400*100.0;
struct ipc_channel { struct ipc_channel {
struct rlist readers, writers; struct rlist readers, writers;
unsigned creaders;
unsigned cwriters;
unsigned size; unsigned size;
unsigned beg; unsigned beg;
unsigned count; unsigned count;
...@@ -73,6 +75,8 @@ void ...@@ -73,6 +75,8 @@ void
ipc_channel_init(struct ipc_channel *ch) ipc_channel_init(struct ipc_channel *ch)
{ {
ch->beg = ch->count = 0; ch->beg = ch->count = 0;
ch->creaders = 0;
ch->cwriters = 0;
rlist_init(&ch->readers); rlist_init(&ch->readers);
rlist_init(&ch->writers); rlist_init(&ch->writers);
} }
...@@ -96,12 +100,13 @@ void * ...@@ -96,12 +100,13 @@ void *
ipc_channel_get_timeout(struct ipc_channel *ch, ev_tstamp timeout) ipc_channel_get_timeout(struct ipc_channel *ch, ev_tstamp timeout)
{ {
/* channel is empty */ /* channel is empty */
if (!ch->count) { if (ch->count == 0 || ch->creaders >= ch->count) {
rlist_add_tail_entry(&ch->readers, fiber, state); rlist_add_tail_entry(&ch->readers, fiber, state);
ch->creaders++;
bool cancellable = fiber_setcancellable(true); bool cancellable = fiber_setcancellable(true);
bool timed_out = fiber_yield_timeout(timeout); bool timed_out = fiber_yield_timeout(timeout);
rlist_del_entry(fiber, state); rlist_del_entry(fiber, state);
ch->creaders--;
fiber_testcancel(); fiber_testcancel();
fiber_setcancellable(cancellable); fiber_setcancellable(cancellable);
...@@ -115,6 +120,7 @@ ipc_channel_get_timeout(struct ipc_channel *ch, ev_tstamp timeout) ...@@ -115,6 +120,7 @@ ipc_channel_get_timeout(struct ipc_channel *ch, ev_tstamp timeout)
} }
} }
assert(ch->count > 0);
void *res = ch->item[ch->beg]; void *res = ch->item[ch->beg];
if (++ch->beg >= ch->size) if (++ch->beg >= ch->size)
ch->beg -= ch->size; ch->beg -= ch->size;
...@@ -127,6 +133,7 @@ ipc_channel_get_timeout(struct ipc_channel *ch, ev_tstamp timeout) ...@@ -127,6 +133,7 @@ ipc_channel_get_timeout(struct ipc_channel *ch, ev_tstamp timeout)
fiber_wakeup(f); fiber_wakeup(f);
} }
return res; return res;
} }
...@@ -141,13 +148,15 @@ ipc_channel_put_timeout(struct ipc_channel *ch, void *data, ...@@ -141,13 +148,15 @@ ipc_channel_put_timeout(struct ipc_channel *ch, void *data,
ev_tstamp timeout) ev_tstamp timeout)
{ {
/* channel is full */ /* channel is full */
if (ch->count >= ch->size) { if (ch->count >= ch->size || ch->cwriters >= ch->size - ch->count) {
rlist_add_tail_entry(&ch->writers, fiber, state); rlist_add_tail_entry(&ch->writers, fiber, state);
ch->cwriters++;
bool cancellable = fiber_setcancellable(true); bool cancellable = fiber_setcancellable(true);
bool timed_out = fiber_yield_timeout(timeout); bool timed_out = fiber_yield_timeout(timeout);
rlist_del_entry(fiber, state); rlist_del_entry(fiber, state);
ch->cwriters--;
fiber_testcancel(); fiber_testcancel();
fiber_setcancellable(cancellable); fiber_setcancellable(cancellable);
...@@ -156,9 +165,10 @@ ipc_channel_put_timeout(struct ipc_channel *ch, void *data, ...@@ -156,9 +165,10 @@ ipc_channel_put_timeout(struct ipc_channel *ch, void *data,
errno = ETIMEDOUT; errno = ETIMEDOUT;
return -1; return -1;
} }
} }
assert(ch->count < ch->size);
unsigned i = ch->beg; unsigned i = ch->beg;
i += ch->count; i += ch->count;
ch->count++; ch->count++;
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment