{"action":"created","issue":{"url":"https://api.github.com/repos/flux-framework/flux-core/issues/225","labels_url":"https://api.github.com/repos/flux-framework/flux-core/issues/225/labels{/name}","comments_url":"https://api.github.com/repos/flux-framework/flux-core/issues/225/comments","events_url":"https://api.github.com/repos/flux-framework/flux-core/issues/225/events","html_url":"https://github.com/flux-framework/flux-core/pull/225","id":90520940,"number":225,"title":"revise reactor API, refactor reactor internals","user":{"login":"garlick","id":169947,"avatar_url":"https://avatars.githubusercontent.com/u/169947?v=3","gravatar_id":"","url":"https://api.github.com/users/garlick","html_url":"https://github.com/garlick","followers_url":"https://api.github.com/users/garlick/followers","following_url":"https://api.github.com/users/garlick/following{/other_user}","gists_url":"https://api.github.com/users/garlick/gists{/gist_id}","starred_url":"https://api.github.com/users/garlick/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/garlick/subscriptions","organizations_url":"https://api.github.com/users/garlick/orgs","repos_url":"https://api.github.com/users/garlick/repos","events_url":"https://api.github.com/users/garlick/events{/privacy}","received_events_url":"https://api.github.com/users/garlick/received_events","type":"User","site_admin":false},"labels":[{"url":"https://api.github.com/repos/flux-framework/flux-core/labels/review","name":"review","color":"ededed"}],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":9,"created_at":"2015-06-23T23:51:38Z","updated_at":"2015-06-25T22:49:26Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/flux-framework/flux-core/pulls/225","html_url":"https://github.com/flux-framework/flux-core/pull/225","diff_url":"https://github.com/flux-framework/flux-core/pull/225.diff","patch_url":"https://github.com/flux-framework/flux-core/pull/225.patch"},"body":"This PR works over the Flux reactor.  I think this is not quite done but wanted to get any feedback started while I continued to work on it.\r\n\r\nThe most recent reactor API interfaces discussed in PR #168 are implemented, leaving the old interfaces available in \"libcompat\", which is linked into libflux-internal.  After the api is stablized we can convert everything and get rid of libcompat.  Here is a summary of the API changes, followed by a description of some cleanup of the internals.\r\n\r\n**general reactor stuff**\r\nOur own POLL bits instead of co-opting zeromq's.   Note that only message handlers are optionally started as co-processes.  This should probably be fixed but I'm not sure I want to do it here as it will be a fairly painful change and this PR is already pretty big.\r\n```C\r\nenum {\r\n    FLUX_POLLIN = 1,\r\n    FLUX_POLLOUT = 2,\r\n    FLUX_POLLERR = 4,\r\n};\r\n\r\nint flux_reactor_start (flux_t h);\r\nvoid flux_reactor_stop (flux_t h);\r\nvoid flux_reactor_stop_error (flux_t h);\r\n\r\nint flux_sleep_on (flux_t h, struct flux_match match);\r\n```\r\n\r\n**message watchers**\r\nThese look like reactor callbacks but they really are message dispatch callbacks.  Do we need to expose a way for an application using an external reactor to instantiate a dispatcher and drive it with messages?\r\n```C\r\ntypedef struct flux_msg_watcher flux_msg_watcher_t;\r\ntypedef void (*flux_msg_watcher_f)(flux_t h, flux_msg_watcher_t *w,\r\n                                   const flux_msg_t *msg, void *arg);\r\n\r\nint flux_msg_watcher_add (flux_t h, struct flux_match match,\r\n                          flux_msg_watcher_f cb, void *arg,\r\n                          flux_msg_watcher_t **wp);\r\nvoid flux_msg_watcher_cancel (flux_msg_watcher_t *w);\r\n\r\nstruct flux_msghandler {\r\n    int typemask;\r\n    char *topic_glob;\r\n    flux_msg_watcher_f cb;\r\n};\r\n#define FLUX_MSGHANDLER_TABLE_END { 0, NULL, NULL }\r\nint flux_msg_watcher_addvec (flux_t h, struct flux_msghandler htab[], void *arg);\r\n```\r\n\r\n**file descriptor watchers**\r\n```C\r\ntypedef struct flux_fd_watcher flux_fd_watcher_t;\r\ntypedef void (*flux_fd_watcher_f)(flux_t h, flux_fd_watcher_t *w,\r\n                                  int fd, int revents, void *arg);\r\nint flux_fd_watcher_add (flux_t h, int fd, int events,\r\n                         flux_fd_watcher_f cb, void *arg,\r\n                         flux_fd_watcher_t **wp);\r\nvoid flux_fd_watcher_cancel (flux_fd_watcher_t *w);\r\n```\r\n\r\n**zeromq socket watchers**\r\nI guess zeromq is not entirely hidden in our API if we're exporting this, although it doesn't require any zeromq/czmq headers.\r\n```C\r\ntypedef struct flux_zmq_watcher flux_zmq_watcher_t;\r\ntypedef void (*flux_zmq_watcher_f)(flux_t h, flux_zmq_watcher_t *w,\r\n                                   void *zsock, int revents, void *arg);\r\nint flux_zmq_watcher_add (flux_t h, void *zsock, int events,\r\n                          flux_zmq_watcher_f cb, void *arg,\r\n                          flux_zmq_watcher_t **wp);\r\nvoid flux_zmq_watcher_cancel (flux_zmq_watcher_t *w);\r\n```\r\n\r\n**timer watchers**\r\nA one shot timer is indicated by setting `repeat` to zero.  There is no way to enable/disable a timer without destroying it and that feels a bit awkward.\r\n```C\r\ntypedef struct flux_timer_watcher flux_timer_watcher_t;\r\ntypedef void (*flux_timer_watcher_f)(flux_t h, flux_timer_watcher_t *w,\r\n                                     int revents, void *arg);\r\nint flux_timer_watcher_add (flux_t h, double after, double repeat,\r\n                            flux_timer_watcher_f cb, void *arg,\r\n                            flux_timer_watcher_t **wp);\r\nvoid flux_timer_watcher_cancel (flux_timer_watcher_t *w);\r\n```\r\n\r\n The reactor was re-factored as discussed in issue #124.  Instead of duplicating a libev loop and re-queue queue in each connector and exposing a large number of connector methods, an edge-triggered pollfd/pollevents interface is exported by each connector and there is only one re-queue queue in the handle with its own pollfd/pollevents interface.   The handle wraps the connector and the re-queue queue using epoll(2) and exposes the result as `flux_pollfd()` and `flux_pollevents()`. These functions are used by the internal libev reactor (using a \"composite watcher\" `src/libflux/ev_flux.[ch]`) and could be used to integrate a flux handle into an external reactor. In fact the internal reactor uses only this interface and no longer directly calls any connector methods which is a sort of \"sufficiency proof\" at least for libev.\r\n\r\nThe specialized handle implementation used by comms modules running in the same address space as the broker was converted to a regular connector called `shmem.so`.  `flux_open()` was not a sufficient interface to pass in the `zctx_t` needed to open the other end of the socket, so a new mode was added, where `flux_open()` is called with the `FLUX_O_NOCONNECT` flag, and a separate `flux_connect()` call connects the socket, after setters are called on the handle.  Rather than implement new setters, the generic `flux_aux_set()` / `flux_aux_get()` functions were used.  Keys used internally by libflux are prefixed with `flux::` to avoid collisions with user keys.  This mode may also prove useful for sharing security contexts with a connector in the future."},"comment":{"url":"https://api.github.com/repos/flux-framework/flux-core/issues/comments/115422852","html_url":"https://github.com/flux-framework/flux-core/pull/225#issuecomment-115422852","issue_url":"https://api.github.com/repos/flux-framework/flux-core/issues/225","id":115422852,"user":{"login":"grondo","id":741970,"avatar_url":"https://avatars.githubusercontent.com/u/741970?v=3","gravatar_id":"","url":"https://api.github.com/users/grondo","html_url":"https://github.com/grondo","followers_url":"https://api.github.com/users/grondo/followers","following_url":"https://api.github.com/users/grondo/following{/other_user}","gists_url":"https://api.github.com/users/grondo/gists{/gist_id}","starred_url":"https://api.github.com/users/grondo/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/grondo/subscriptions","organizations_url":"https://api.github.com/users/grondo/orgs","repos_url":"https://api.github.com/users/grondo/repos","events_url":"https://api.github.com/users/grondo/events{/privacy}","received_events_url":"https://api.github.com/users/grondo/received_events","type":"User","site_admin":false},"created_at":"2015-06-25T22:49:26Z","updated_at":"2015-06-25T22:49:26Z","body":"Also @garlick -- I see your point about `watcher_add` pattern vs `watcher_create` and `watcher_start`. \r\n\r\nI don't have any good advice here, except that I also prefer the second pattern (create and start), even though it is more verbose. However, your point that it doesn't fit the message dispatcher paradigm is germane. It feels like maybe if we could tease apart the message dispatcher vs message watcher,  this mismatch could be partially solved? (Sorry I don't have any better thoughts at this time)\r\n\r\n"}}