-
Notifications
You must be signed in to change notification settings - Fork 484
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Solution: port CLIENT and SERVER sockets from libzmq, which are thread-safe sockets. The thread-safe sockets family includes other sockets, which are not part of this commit. Also, CLIENT and SERVER cannot be used with a poller at the moment. For poller support, zmq_poller from libzmq has to be ported as well.
- Loading branch information
Showing
14 changed files
with
1,399 additions
and
641 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,8 @@ | ||
package zmq; | ||
|
||
import java.io.Closeable; | ||
|
||
public interface IMailbox extends Closeable { | ||
void send(final Command cmd); | ||
Command recv(long timeout); | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,129 @@ | ||
package zmq; | ||
|
||
import zmq.pipe.YPipe; | ||
import zmq.util.Errno; | ||
|
||
import java.io.IOException; | ||
import java.util.Vector; | ||
import java.util.concurrent.TimeUnit; | ||
import java.util.concurrent.locks.ReentrantLock; | ||
import java.util.concurrent.locks.Condition; | ||
|
||
public class MailboxSafe implements IMailbox | ||
{ | ||
// The pipe to store actual commands. | ||
private final YPipe<Command> cpipe; | ||
|
||
// Synchronize access to the mailbox from receivers and senders | ||
private final ReentrantLock sync; | ||
|
||
// Condition variable to pass signals from writer thread to reader thread. | ||
private final Condition condition; | ||
|
||
private final Vector<Signaler> signalers; | ||
|
||
// mailbox name, for better debugging | ||
private final String name; | ||
|
||
private final Errno errno; | ||
|
||
public MailboxSafe(Ctx ctx, ReentrantLock sync, String name) | ||
{ | ||
this.errno = ctx.errno(); | ||
this.cpipe = new YPipe<>(Config.COMMAND_PIPE_GRANULARITY.getValue()); | ||
this.sync = sync; | ||
this.condition = this.sync.newCondition(); | ||
this.signalers = new Vector<Signaler>(10); | ||
this.name = name; | ||
|
||
// Get the pipe into passive state. That way, if the users starts by | ||
// polling on the associated file descriptor it will get woken up when | ||
// new command is posted. | ||
Command cmd = cpipe.read(); | ||
assert (cmd == null); | ||
} | ||
|
||
public void addSignaler(Signaler signaler) | ||
{ | ||
this.signalers.add(signaler); | ||
} | ||
|
||
public void removeSignaler(Signaler signaler) | ||
{ | ||
this.signalers.remove(signaler); | ||
} | ||
|
||
public void clearSignalers() | ||
{ | ||
this.signalers.clear(); | ||
} | ||
|
||
@Override | ||
public void send(Command cmd) | ||
{ | ||
sync.lock(); | ||
try { | ||
cpipe.write(cmd, false); | ||
boolean ok = cpipe.flush(); | ||
|
||
if (!ok) { | ||
condition.signalAll(); | ||
signalers.forEach(Signaler::send); | ||
} | ||
} | ||
finally { | ||
sync.unlock(); | ||
} | ||
} | ||
|
||
@Override | ||
public Command recv(long timeout) | ||
{ | ||
Command cmd; | ||
|
||
// Try to get the command straight away. | ||
cmd = cpipe.read(); | ||
if (cmd != null) { | ||
return cmd; | ||
} | ||
|
||
// If the timeout is zero, it will be quicker to release the lock, giving other a chance to send a command | ||
// and immediately relock it. | ||
if (timeout == 0) { | ||
sync.unlock(); | ||
sync.lock(); | ||
} else { | ||
try { | ||
// Wait for signal from the command sender. | ||
condition.await(timeout, TimeUnit.MILLISECONDS); | ||
} catch (InterruptedException e) { | ||
errno.set(ZError.EINTR); | ||
return null; | ||
} | ||
} | ||
|
||
// Another thread may already fetch the command | ||
cmd = cpipe.read(); | ||
if (cmd == null) { | ||
errno.set(ZError.EAGAIN); | ||
return null; | ||
} | ||
|
||
return cmd; | ||
} | ||
|
||
@Override | ||
public void close() throws IOException | ||
{ | ||
// Work around problem that other threads might still be in our | ||
// send() method, by waiting on the mutex before disappearing. | ||
sync.lock(); | ||
sync.unlock(); | ||
} | ||
|
||
@Override | ||
public String toString() | ||
{ | ||
return super.toString() + "[" + name + "]"; | ||
} | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.