Built-in map-reduce utils

[Khatskevich]

Provide an Api which executes some stored procedure on each replicaset.
It is especially useful when a router in working as a proxy server and an application cannot access replicaset map.

[Gerold103]

Actually, it is a more common task, called map-reduce. It is also required by @racktear.

[Gerold103]

It was decided we need a general map-reduce tool to do consistent map-reduces. Their main application is going to be consistent DQL in cluster, to select tuples from multiple storages.

On the other hand, this must not depend on schema, because DQL is only one of possible applications.

Consistent means that during the map-reduce call all the not scanned buckets should stay in place. It should not be possible, that a bucket is not covered by a map-reduce or covered twice. For example, a bucket could be stored on a not yet scanned storage1. Then it is moved to already scanned storage2 before storage1 is scanned too. So it won't be visible to the on-going map-reduce request.

That means the map-reduce calls conflict with rebalancer - rebalancing and map-reduce can't be done simultaneously. Blocking of one of them in favor of another would lead to starvation. So there must be some kind of fair switching, when rebalancer and map-reduce requests are handled both.

A possible implementation idea - implement pin-map-reduce.

This is how rebalancer will work:

Each storage has a counter - number of currently running map-reduce calls on the storage. If the counter is 0, and a rebalancer request arrives, it is applied.

If the counter is not zero, the rebalancer request is queued, and new map-reduce requests are either rejected or stacked until the rebalancer request is finished.

If on the storage there are no map-reduce calls, it is also necessary to ensure there are no map-reduces on the receiving side.

This is how map-reduce calls will work:

Each call has 3 phases: pin, map, reduce. During pin the router will go to all storages and try to create a uniquely identified pin-map-reduce (PMR) object with a timeout. The object increments number of currently running map-reduce calls on the storage, and starts a timeout, after which it will delete itself.

The increment works only if there are no running or queued rebalancer requests. Otherwise an error is returned, or the call is queued after the rebalancer requests.

Router gets a unique ID from each storage. Kind of a session ID, a cookie. If each storage returned an ID, the router may be sure it can try to send map calls to each storage. Each call should be accompanied by the ID returned from that storage.

On the storage the call will check if the PMR object with the given ID still exists. If it doesn't, it means it was expired already. Need to retry everything from the scratch.

If there is a PMR with the given ID, its timeout is set to infinity so as not to start rebalancing during map execution, and the map is executed on this storage. After that the PMR is deleted.

Router collects responses. If every map returned success, the pin-map stages are finished. The reduce stage is on the client, so we don't care about this phase.

With this algorithm it seems both map-reduce and rebalancing requests are going to be executed eventually.

As an optimization, we can consider not forbidding to send and receive buckets, but only forbid to activate the received buckets, and to garbage collect the sent buckets. But that would make it harder for the application code, as it will need to check each tuple if its bucket is accessible on the storage.