Pyzmq's eventloop.future context with 'inproc' in worker threads
.everyoneloves__top-leaderboard:empty,.everyoneloves__mid-leaderboard:empty,.everyoneloves__bot-mid-leaderboard:empty{ height:90px;width:728px;box-sizing:border-box;
}
Part of my application consists of a main thread and a couple of worker threads that send their results to the main thread using pyzmq sockets. The main thread is running a tornado IOloop and uses async functions to read the incoming data on a variety of socket types created using a future.context.
For performance reasons I would like to use the inproc protocol. However inproc only works when the main thread and the worker threads share the identical context. This on the other hand would require that each worker thread needs to run a tornado IOloop, which I think is bit a overkill for simple workers.
A minimal example to illustrate the problem:
import time
from threading import Thread
import zmq
from zmq.eventloop.future import Context as FutureContext
import tornado.ioloop
def worker(ctx):
socket = ctx.socket(zmq.PUSH)
socket.bind('inproc://worker')
while True:
# Do some work
time.sleep(2)
socket.send_pyobj("Work done")
async def mainLoop(ctx):
socket = ctx.socket(zmq.PULL)
socket.connect('inproc://worker')
while True:
#print(socket.recv_pyobj())
print(await socket.recv_pyobj())
normalCtx = zmq.Context()
futureCtx = FutureContext()
t = Thread(target=worker, kwargs=dict(ctx=normalCtx))
t.start()
# wait for bind to be effective
time.sleep(4)
io_loop = tornado.ioloop.IOLoop.current()
io_loop.spawn_callback(mainLoop, ctx=futureCtx)
io_loop.start()
In the example, the PULL socket won't receive messages since it is issued on a different context than the worker socket. The example works fine if I use the normalCtx in both threads (removing await). It also works fine when using TCP as the transport protocol.
The solutions to get it working I found are:
- Use
normalcontext and give up theasync/await. - Use
futurecontext and run anioloopin each worker. - Use TCP as the transport protocol.
My question is if there is a magic trick to get it working with async/await, inproc and not having to run ioloops in the workers, e.g. by accessing the future context in a non-future way?
multithreading async-await tornado pyzmq inproc
edited Nov 21 '18 at 20:56
Benyamin Jafari
3,72352453
asked Oct 15 '18 at 19:33
SchwingkopfSchwingkopf
183
add a comment |
Part of my application consists of a main thread and a couple of worker threads that send their results to the main thread using pyzmq sockets. The main thread is running a tornado IOloop and uses async functions to read the incoming data on a variety of socket types created using a future.context.
For performance reasons I would like to use the inproc protocol. However inproc only works when the main thread and the worker threads share the identical context. This on the other hand would require that each worker thread needs to run a tornado IOloop, which I think is bit a overkill for simple workers.
A minimal example to illustrate the problem:
import time
from threading import Thread
import zmq
from zmq.eventloop.future import Context as FutureContext
import tornado.ioloop
def worker(ctx):
socket = ctx.socket(zmq.PUSH)
socket.bind('inproc://worker')
while True:
# Do some work
time.sleep(2)
socket.send_pyobj("Work done")
async def mainLoop(ctx):
socket = ctx.socket(zmq.PULL)
socket.connect('inproc://worker')
while True:
#print(socket.recv_pyobj())
print(await socket.recv_pyobj())
normalCtx = zmq.Context()
futureCtx = FutureContext()
t = Thread(target=worker, kwargs=dict(ctx=normalCtx))
t.start()
# wait for bind to be effective
time.sleep(4)
io_loop = tornado.ioloop.IOLoop.current()
io_loop.spawn_callback(mainLoop, ctx=futureCtx)
io_loop.start()
In the example, the PULL socket won't receive messages since it is issued on a different context than the worker socket. The example works fine if I use the normalCtx in both threads (removing await). It also works fine when using TCP as the transport protocol.
The solutions to get it working I found are:
- Use
normalcontext and give up theasync/await. - Use
futurecontext and run anioloopin each worker. - Use TCP as the transport protocol.
My question is if there is a magic trick to get it working with async/await, inproc and not having to run ioloops in the workers, e.g. by accessing the future context in a non-future way?
multithreading async-await tornado pyzmq inproc
edited Nov 21 '18 at 20:56
Benyamin Jafari
3,72352453
asked Oct 15 '18 at 19:33
SchwingkopfSchwingkopf
183
add a comment |
Part of my application consists of a main thread and a couple of worker threads that send their results to the main thread using pyzmq sockets. The main thread is running a tornado IOloop and uses async functions to read the incoming data on a variety of socket types created using a future.context.
For performance reasons I would like to use the inproc protocol. However inproc only works when the main thread and the worker threads share the identical context. This on the other hand would require that each worker thread needs to run a tornado IOloop, which I think is bit a overkill for simple workers.
A minimal example to illustrate the problem:
import time
from threading import Thread
import zmq
from zmq.eventloop.future import Context as FutureContext
import tornado.ioloop
def worker(ctx):
socket = ctx.socket(zmq.PUSH)
socket.bind('inproc://worker')
while True:
# Do some work
time.sleep(2)
socket.send_pyobj("Work done")
async def mainLoop(ctx):
socket = ctx.socket(zmq.PULL)
socket.connect('inproc://worker')
while True:
#print(socket.recv_pyobj())
print(await socket.recv_pyobj())
normalCtx = zmq.Context()
futureCtx = FutureContext()
t = Thread(target=worker, kwargs=dict(ctx=normalCtx))
t.start()
# wait for bind to be effective
time.sleep(4)
io_loop = tornado.ioloop.IOLoop.current()
io_loop.spawn_callback(mainLoop, ctx=futureCtx)
io_loop.start()
In the example, the PULL socket won't receive messages since it is issued on a different context than the worker socket. The example works fine if I use the normalCtx in both threads (removing await). It also works fine when using TCP as the transport protocol.
The solutions to get it working I found are:
- Use
normalcontext and give up theasync/await. - Use
futurecontext and run anioloopin each worker. - Use TCP as the transport protocol.
My question is if there is a magic trick to get it working with async/await, inproc and not having to run ioloops in the workers, e.g. by accessing the future context in a non-future way?
multithreading async-await tornado pyzmq inproc
edited Nov 21 '18 at 20:56
Benyamin Jafari
3,72352453
asked Oct 15 '18 at 19:33
SchwingkopfSchwingkopf
183
Part of my application consists of a main thread and a couple of worker threads that send their results to the main thread using pyzmq sockets. The main thread is running a tornado IOloop and uses async functions to read the incoming data on a variety of socket types created using a future.context.
For performance reasons I would like to use the inproc protocol. However inproc only works when the main thread and the worker threads share the identical context. This on the other hand would require that each worker thread needs to run a tornado IOloop, which I think is bit a overkill for simple workers.
A minimal example to illustrate the problem:
import time
from threading import Thread
import zmq
from zmq.eventloop.future import Context as FutureContext
import tornado.ioloop
def worker(ctx):
socket = ctx.socket(zmq.PUSH)
socket.bind('inproc://worker')
while True:
# Do some work
time.sleep(2)
socket.send_pyobj("Work done")
async def mainLoop(ctx):
socket = ctx.socket(zmq.PULL)
socket.connect('inproc://worker')
while True:
#print(socket.recv_pyobj())
print(await socket.recv_pyobj())
normalCtx = zmq.Context()
futureCtx = FutureContext()
t = Thread(target=worker, kwargs=dict(ctx=normalCtx))
t.start()
# wait for bind to be effective
time.sleep(4)
io_loop = tornado.ioloop.IOLoop.current()
io_loop.spawn_callback(mainLoop, ctx=futureCtx)
io_loop.start()
In the example, the PULL socket won't receive messages since it is issued on a different context than the worker socket. The example works fine if I use the normalCtx in both threads (removing await). It also works fine when using TCP as the transport protocol.
The solutions to get it working I found are:
- Use
normalcontext and give up theasync/await. - Use
futurecontext and run anioloopin each worker. - Use TCP as the transport protocol.
My question is if there is a magic trick to get it working with async/await, inproc and not having to run ioloops in the workers, e.g. by accessing the future context in a non-future way?
multithreading async-await tornado pyzmq inproc
multithreading async-await tornado pyzmq inproc
edited Nov 21 '18 at 20:56
Benyamin Jafari
3,72352453
asked Oct 15 '18 at 19:33
SchwingkopfSchwingkopf
183
edited Nov 21 '18 at 20:56
Benyamin Jafari
3,72352453
asked Oct 15 '18 at 19:33
SchwingkopfSchwingkopf
183
edited Nov 21 '18 at 20:56
Benyamin Jafari
3,72352453
edited Nov 21 '18 at 20:56
Benyamin Jafari
3,72352453
edited Nov 21 '18 at 20:56
Benyamin Jafari
3,72352453
3,72352453
asked Oct 15 '18 at 19:33
SchwingkopfSchwingkopf
183
asked Oct 15 '18 at 19:33
SchwingkopfSchwingkopf
183
asked Oct 15 '18 at 19:33
SchwingkopfSchwingkopf
183
183
add a comment |
add a comment |
0
active
oldest
votes
Your Answer
StackExchange.ifUsing("editor", function () {
StackExchange.using("externalEditor", function () {
StackExchange.using("snippets", function () {
StackExchange.snippets.init();
});
});
}, "code-snippets");
StackExchange.ready(function() {
var channelOptions = {
tags: "".split(" "),
id: "1"
};
initTagRenderer("".split(" "), "".split(" "), channelOptions);
StackExchange.using("externalEditor", function() {
// Have to fire editor after snippets, if snippets enabled
if (StackExchange.settings.snippets.snippetsEnabled) {
StackExchange.using("snippets", function() {
createEditor();
});
}
else {
createEditor();
}
});
function createEditor() {
StackExchange.prepareEditor({
heartbeatType: 'answer',
autoActivateHeartbeat: false,
convertImagesToLinks: true,
noModals: true,
showLowRepImageUploadWarning: true,
reputationToPostImages: 10,
bindNavPrevention: true,
postfix: "",
imageUploader: {
brandingHtml: "Powered by u003ca class="icon-imgur-white" href="https://imgur.com/"u003eu003c/au003e",
contentPolicyHtml: "User contributions licensed under u003ca href="https://creativecommons.org/licenses/by-sa/3.0/"u003ecc by-sa 3.0 with attribution requiredu003c/au003e u003ca href="https://stackoverflow.com/legal/content-policy"u003e(content policy)u003c/au003e",
allowUrls: true
},
onDemand: true,
discardSelector: ".discard-answer"
,immediatelyShowMarkdownHelp:true
});
}
});
Sign up or log in
StackExchange.ready(function () {
StackExchange.helpers.onClickDraftSave('#login-link');
});
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Post as a guest
Required, but never shown
StackExchange.ready(
function () {
StackExchange.openid.initPostLogin('.new-post-login', 'https%3a%2f%2fstackoverflow.com%2fquestions%2f52823585%2fpyzmqs-eventloop-future-context-with-inproc-in-worker-threads%23new-answer', 'question_page');
}
);
Post as a guest
Required, but never shown
0
active
oldest
votes
0
active
oldest
votes
active
oldest
votes
active
oldest
votes
Thanks for contributing an answer to Stack Overflow!
- Please be sure to answer the question. Provide details and share your research!
But avoid …
- Asking for help, clarification, or responding to other answers.
- Making statements based on opinion; back them up with references or personal experience.
To learn more, see our tips on writing great answers.
Sign up or log in
StackExchange.ready(function () {
StackExchange.helpers.onClickDraftSave('#login-link');
});
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Post as a guest
Required, but never shown
StackExchange.ready(
function () {
StackExchange.openid.initPostLogin('.new-post-login', 'https%3a%2f%2fstackoverflow.com%2fquestions%2f52823585%2fpyzmqs-eventloop-future-context-with-inproc-in-worker-threads%23new-answer', 'question_page');
}
);
Post as a guest
Required, but never shown
Sign up or log in
StackExchange.ready(function () {
StackExchange.helpers.onClickDraftSave('#login-link');
});
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Post as a guest
Required, but never shown
Sign up or log in
StackExchange.ready(function () {
StackExchange.helpers.onClickDraftSave('#login-link');
});
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Post as a guest
Required, but never shown
Sign up or log in
StackExchange.ready(function () {
StackExchange.helpers.onClickDraftSave('#login-link');
});
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Post as a guest
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown
