{"action":"created","issue":{"url":"https://api.github.com/repos/flux-framework/flux-core/issues/785","repository_url":"https://api.github.com/repos/flux-framework/flux-core","labels_url":"https://api.github.com/repos/flux-framework/flux-core/issues/785/labels{/name}","comments_url":"https://api.github.com/repos/flux-framework/flux-core/issues/785/comments","events_url":"https://api.github.com/repos/flux-framework/flux-core/issues/785/events","html_url":"https://github.com/flux-framework/flux-core/issues/785","id":173508797,"number":785,"title":"Questions about reduction handle (flux_reduce_t)","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},"labels":[{"url":"https://api.github.com/repos/flux-framework/flux-core/labels/question","name":"question","color":"cc317c"}],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":19,"created_at":"2016-08-26T17:59:02Z","updated_at":"2016-09-06T19:58:25Z","closed_at":null,"body":"I'm currently experimenting with `flux_reduce_t` and finding it a bit awkward for my use case, however I assume I'm just misusing or do not understand how the interface should be used, so I figured I'd open an issue and get advice.\r\n\r\nMy use case is an \"aggregator\" module which offers a service to aggregate values from integer source IDs (e.g. rank or taskid) into a kvs key. At this point the idea is experimental but will eventually be used to aggregate task exit status to a single kvs key/directory using reduction over the TBON.\r\n\r\nRepresentation of an aggregate at this point would look something like\r\n```\r\n# jcr-version 0.6\r\n\r\naggregate_entry {\r\n  \"ids\"   :string,  ; nodeset\r\n  \"value\" :integer\r\n}\r\n\r\naggregate {\r\n  \"key\"   :string,  ; destination kvs key\r\n  \"total\" :integer, ; total values to aggregate \r\n  \"count\" :integer, ; count of entries in this aggregate\r\n  \"min\"   :integer,\r\n  \"max\"   :integer,\r\n  \"entries\" :[ *: aggregate_entry ]\r\n}\r\n```\r\n\r\nI'm not sure yet the best way to push the fully reduced object to the kvs, but that could be determined later.\r\n\r\nThe concept of how this will work is that on task exit, each `wrexecd` will push the task exit message like (for jobid 13 with 1K tasks)\r\n\r\n```json\r\n{\r\n  \"key\" : \"lwj.13.exit_status\",\r\n  \"total\": 1024,\r\n  \"count\": 1,\r\n  \"entries\" : [\r\n    {  \"ids\" : \"0\",\r\n       \"value\" : 0\r\n    } ]\r\n}\r\n```\r\n\r\nWhen the `aggregator` module receives the first message above on a rank, it will create\r\na new reduction handle and some state and hash the state on the `\"key\"`.\r\n\r\nIncoming messages can be reduced by finding matching values in the entries list and merging the \"ids\" nodesets and updating count.\r\n\r\nRank 0 will `sink` the data to the kvs once `count == total`.\r\n\r\nThere are just a couple of awkward points currently:\r\n\r\n * For the specific case of the task exit status, we'd like to do a perfect reduction and wait for all local and upstream aggregate messages before forwarding, but it isn't actually trivial to calculate how many total items to expect at any given rank (except rank 0) (a job may not be running on all descendant ranks), nor even how many tasks to expect from each rank. (I imagine this one is solvable, but for now I supposed timed flush should be better than nothing)\r\n\r\n * The queue semantics of the reduction handle seem to make processing these aggregate messages a little bit awkward. My initial approach was to process the `aggregate.push` messages into C data structures hashed by the requested kvs key. The straightforward way to do this would be to directly \"merge\" the incoming JSON message with an existing object, or if \"key\" does not yet exist create one. Then when reduction timeout or HWM is met, convert the current aggregate to JSON and forward it upstream. However, since the reduction handle requires using its queue, incoming messages have to be appended to the queue, then popped off and merged. Because you have to push something back onto the reduction handle, though, you have to push something back in the same form -- this requires either converting back to JSON, or translating all incoming requests to the internal aggregate type first, then pushing on the list. Either way there is some extra code required to use the `flux_reduce_t`.\r\n \r\n Would it make sense to have a mode of `flux_reduce_t` where it doesn't implement queue interface,   and instead you update the reduction handle with counts (`flux_reduce_increment ()`, `flux_reduce_decrement ()`) and it uses the explicit count to determine when to call flush? Perhaps the timer/HWM code could be abstracted into a simpler interface, which would then be used to create other reduction interfaces -- e..g the current could be renamed `flux_reduce_queue_t`?\r\n\r\n* The other confusing item is the `batch` sequence number. I'm assuming that does not apply to this specific case, where we are creating reduction handles on the fly and destroying them, and `batch` can safely be set to `0`.?\r\n\r\nAgain, I might be having these problems only because I'm using the interface incorrectly, so please bear with me! Hopefully any advice in this issue can also assist others."},"comment":{"url":"https://api.github.com/repos/flux-framework/flux-core/issues/comments/245070998","html_url":"https://github.com/flux-framework/flux-core/issues/785#issuecomment-245070998","issue_url":"https://api.github.com/repos/flux-framework/flux-core/issues/785","id":245070998,"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":"2016-09-06T19:58:25Z","updated_at":"2016-09-06T19:58:25Z","body":"> You could also just use the KVS and then you'll have a namespace to play with - otherwise I guess you'd need to map your own names to blobrefs and keep that namespace consistent.\r\n\r\nYeah, I did go there first of course, but I thought that was possibly overkill for this use case. The content store seemed like an interesting way to distribute a possibly large payload to all ranks without using the 1:many model of `flux_rpc_multi`, and without storing all state in the KVS (thus increasing the number of kvs_commits)"}}
{"actor":{"display_login":"grondo"}}