Pipe block operators proposal

Block-template operator

• Excellent parallel
  • map(fn)
    • Note you can use map without specifying map function in pipelines.
  • map_join(fn) result must be object. Adds keys to original frame-object and passes to next.
  • filter(pred)
• Good parallel (local accumulators)
  • reduce(fn) (makes sure that size >= 1)
  • reduce0(zero, fn) (also takes zero element. Can accept zero-sized stream)
  • group_by(field) (in fact it's reduce by google whitepaper)
  • group_by(fn)
  • partition_by(pred) (??? должен разделять по условию... формат???)
    • split(object)
    • split<size_of_predicate_window=1,step=1>(pred)
  • reduce_window_sub<size,step=1>(fn,sub_fn) uses subtract function for 2op update of accumulator.
• Bad parallel
  • sort<asc>(field)
  • sort<asc>(field1,field2) field2 used if field1 was equal
  • sort<asc>(fn)
  • unique(field) (alias disctinct)
  • unique(fn)
• Shrinkable
  • skip(count) -> skip<count>
  • skip(pred)
  • limit(count) -> limit<count>
  • limit(pred)
  • any(pred)
  • all(pred)
• Debug
  • peek(dst)
  • print (==peek(stdout))

Frame operators

• pack<3> N frames into 1.
• unpack 'field' 1 frame into N. (in fact combos with pack-blocking)
  • You can add field with map_join function
• window<size,step=1> e.g. pack<3> == window<3,3>
• batch<1s,1000> collect all frames up to first limit: 1 sec or 1000 elements
  • batch<1s>
  • batch<1000> == pack<1000>
• in_frame<size_in,size_out>({frame_producer, pipeline, frame_reducer})
  • can be in_frame<1,1>
  • can be in_frame<1,N> similar to unpack

Bucket operators

• bucket_split({split_key_access_fn, per_bucket_map, per_bucket_reduce, bucket_reduce}) generic case
  • reducer can be simple joiner to 1 frame
  • reducer can pass frames "as is"
  • bucket map can be implemented in per_bucket_reduce. Implementing it at bucket_reduce is not recommended for best latency, because per_bucket_reduce is N parallel operations, bucket_reduce at each stage reduce parallelism twice.
  • This is original map-reduce by google. map should expose key that splits frames into different buckets. Reduce process per bucket. Result is array of reduce results, not scalar.
    • For full compatibility you should use map+unpack, because original map can emit multiple key-value pairs.

Block operators

• Pack each block to make it operational over N-subblock-frame
  • unroll<4>
  • vectorize<4> will vectorize dividable part of unrolled

Alignment

• Array of objects (default)
• Object of arrays
  • Object of mixed arrays
    • Fusing same type into single array
      • pow 2 fusing
    • Partial fusing
    • Rest trash into object
  • How to
    • align<unroll,separate> # Default=object of arrays, may be overriden
      • align_fuse 'field_a', 'field_b' # Force fuse N fields. Transitive. Must be same type in first implementations.
        • align_fuse<4> 'r', 'g', 'b' # Will fill 4th position with 0
      • align_fuse_type 'type_name' # Force fuse for all fields of specified type
      • align_rest 'field_a' # Force field be not unrolled, but putted in "rest" object
    • align<unroll,type_fuse> # Default=object of arrays, all fields of same type fused
    • align<unroll,manual> # Default=all in "rest" object
      • align_unroll 'field' # Remove from rest. Multiple field same as align_fuse. So use 1-liners for separate unroll.
    • align<> # allows autotuner make special alignment for this specific thread (useful for pipe split/join)

Compute unit separator

• barrier Splits manually pipeline into chunks that can be mapped on different compute units
  • barrier<local> only same machine
  • barrier<remote,durable=true> left and right are splitable on different machines
    • barrier<remote,can_drop> don't make retry wrapper. Latency matters, "all frame must be processed" not matters

Throuput control

• fifo can be between machines.
  • wheel_barrier<size> single machine only. Producer/consumer separated by cycle buffer. 2 locks read, write. Limits producer if buffer filled.
• throttle
• busy_drop<timeout> drops frame if timeout occurs on right side. So right side will not be overloaded.
• stop_barrier stop event cause stop frame propagation on this point
  • stop_barrier_fifo will store all frames in fifo during stopping, send on resume
• reject_barrier will not pass frame if frame id was interrupted by sender (stop signal). (Don't waste computation time if task already not needed)
• warm_up({metric,max_acceleration,time_window,deceleration_over_time}) limits rate acceleration (accident rate grow can kill some services)

Wrapper combo

• barrier(pipeline) == | barrier | pipeline | barrier |
• resource_lock(resource_list, pipeline) resource list lock is transaction. Timeout passed as resource.

Black magic ???

• Loop Stream Detector size/format tuning
  • some info
  • intel info
• CPU cache line tuning
  • frame_size<1MB> # Causes autotuner pack block-operations in blob that will precess 1MB data frame per tick. Round down, but at last 1 (will provide warning if constrain not met).
• CPU time quant granularity rounding/tuning (enough work for thread)
  • time_frame<1ms> # similar to frame_size