anchor.diff:  %% Internal functions
   %% ===================================================================
   
  +-spec block_sorter({pos_integer(), term()}, {pos_integer(), term()}) -> boolean().
  +block_sorter({A, _}, {B, _}) ->
  +    A < B.
  +
  +-spec read_blocks(binary(), binary(), binary(), [pid()], pos_integer(), pos_integer()) ->
  +                         {pos_integer(), [pid()]}.
  +read_blocks(Bucket, Key, UUID, FreeReaders, NextBlock, TotalBlocks) ->
  +    read_blocks(Bucket,
  +                binary_to_list(Key),
  +                UUID,
  +                FreeReaders,
  +                NextBlock,
  +                TotalBlocks,
  +                0).
  +
  +-spec read_blocks(binary(),
  +                  binary(),
  +                  binary(),
  +                  [pid()],
  +                  pos_integer(),
  +                  pos_integer(),
  +                  non_neg_integer()) ->
  +                         {pos_integer(), [pid()]}.
  +read_blocks(_Bucket, _Key, _UUID, [], _, _, ReadsRequested) ->
  +    {ReadsRequested, []};
  +read_blocks(_Bucket, _Key, _UUID, FreeReaders, _TotalBlocks, _TotalBlocks, ReadsRequested) ->
  +    {ReadsRequested, FreeReaders};
  +read_blocks(Bucket, Key, UUID, [ReaderPid | RestFreeReaders], NextBlock, _TotalBlocks, ReadsRequested) ->
  +    riak_moss_block_server:get_block(ReaderPid, Bucket, Key, UUID, NextBlock),
  +    read_blocks(Bucket, Key, UUID, RestFreeReaders, NextBlock+1, _TotalBlocks, ReadsRequested+1).
  +
   %% @private
   %% @doc Start a number of riak_moss_block_server processes
   %% and return a list of their pids.
  
positive.code:                 mani_fsm_pid :: pid(),
                  bucket :: term(),
                  key :: term(),
                  value_cache :: binary(),
                  metadata_cache :: term(),
                  chunk_queue :: term(),
                  manifest :: term(),
                  manifest_uuid :: term(),
                  blocks_left :: list(),
                  test=false :: boolean(),
                  free_readers :: [pid()],
                  all_reader_pids :: [pid()]}).
  
  
negative.code:     {stop, normal, State};
  waiting_continue_or_stop(stop, State) ->
      {stop, normal, State};
  waiting_continue_or_stop(continue, #state{manifest=Manifest,
                                            bucket=BucketName,
                                            key=Key,
                                            next_block=NextBlock,
                                            manifest_uuid=UUID,
                                            free_readers=FreeReaders}=State) ->
      BlockSequences = riak_moss_lfs_utils:block_sequences_for_manifest(Manifest),
      case BlockSequences of
          [] ->
              %% No blocks = empty file
              {stop, normal, State};
          [_|_] ->
              BlocksLeft = sets:from_list(BlockSequences),
              TotalBlocks = sets:size(BlocksLeft),
  
              %% start retrieving the first set of blocks
              {LastBlockRequested, UpdFreeReaders} =
                  read_blocks(BucketName, Key, UUID, FreeReaders, NextBlock, TotalBlocks+1),
              NewState = State#state{blocks_left=BlocksLeft,
                                     last_block_requested=LastBlockRequested,
                                     total_blocks=TotalBlocks,
                                     free_readers=UpdFreeReaders},
              {next_state, waiting_chunks, NewState}
      end.
  
  waiting_chunks(get_next_chunk, From, #state{block_buffer=[], from=PreviousFrom}=State) when PreviousFrom =:= undefined ->
      %% we don't have a chunk ready
      %% yet, so we'll make note
      %% of the sender and go back
      %% into waiting for another
      %% chunk
      {next_state, waiting_chunks, State#state{from=From}};
  waiting_chunks(get_next_chunk,
                 _From,
                 State=#state{block_buffer=[{NextBlock, Block} | RestBlockBuffer],
                              next_block=NextBlock}) ->
      {reply, {chunk, Block}, waiting_chunks, State#state{block_buffer=RestBlockBuffer,
                                                          next_block=NextBlock+1}};
  waiting_chunks(get_next_chunk,
                 From,
                 State) ->
      {next_state, waiting_chunks, State#state{from=From}}.
  
  waiting_chunks({chunk, _Pid, {NextBlock, BlockReturnValue}}, #state{from=From,
                                                                     blocks_left=Remaining,
                                                                     manifest_uuid=UUID,
                                                                     key=Key,
                                                                     bucket=BucketName,
                                                                     next_block=NextBlock,
                                                                     free_readers=FreeReaders,
                                                                     last_block_requested=LastBlockRequested,
                                                                     total_blocks=TotalBlocks,
                                                                     block_buffer=BlockBuffer}=State) ->
  
      {ok, BlockValue} = BlockReturnValue,
      NewRemaining = sets:del_element(NextBlock, Remaining),
      BlocksLeft = sets:size(NewRemaining),
      case From of
          undefined ->
              UpdBlockBuffer =
                  lists:sort(fun block_sorter/2,
                             [{NextBlock, BlockValue} | BlockBuffer]),
              NewState0 = State#state{blocks_left=NewRemaining,
                                     block_buffer=UpdBlockBuffer,
                                     from=undefined},
              case BlocksLeft of
                  0 ->
                      NewState=NewState0,
                      NextStateName = sending_remaining;
                  _ ->
                      {ReadRequests, UpdFreeReaders} =
                          read_blocks(BucketName, Key, UUID, FreeReaders, NextBlock, TotalBlocks),
                      NewState = NewState0#state{last_block_requested=LastBlockRequested+ReadRequests,
                                                 free_readers=UpdFreeReaders},
                      NextStateName = waiting_chunks
              end,
              {next_state, NextStateName, NewState};
          _ ->
              NewState0 = State#state{blocks_left=NewRemaining,
                                     from=undefined},
              case BlocksLeft of
                  0 ->
                      gen_fsm:reply(From, {done, BlockValue}),
                      NewState=NewState0,
                      {stop, normal, NewState};
                  _ ->
                      gen_fsm:reply(From, {chunk, BlockValue}),
                      {ReadRequests, UpdFreeReaders} =
                          read_blocks(BucketName, Key, UUID, FreeReaders, LastBlockRequested+1, TotalBlocks),
                      NewState = NewState0#state{last_block_requested=LastBlockRequested+ReadRequests,
                                                 free_readers=UpdFreeReaders},
                      {next_state, waiting_chunks, NewState}
              end
      end;
  
  waiting_chunks({chunk, _Pid, {BlockSeq, BlockReturnValue}}, #state{blocks_left=Remaining,
                                                                     manifest_uuid=UUID,
                                                                     key=Key,
                                                                     bucket=BucketName,
                                                                     free_readers=FreeReaders,
                                                                     last_block_requested=LastBlockRequested,
                                                                     total_blocks=TotalBlocks,
                                                                     block_buffer=BlockBuffer}=State) ->
      %% we don't deal with missing chunks
      %% at all here, so this pattern
      %% match will fail
      {ok, BlockValue} = BlockReturnValue,
      NewRemaining = sets:del_element(BlockSeq, Remaining),
      BlocksLeft = sets:size(NewRemaining),
      UpdBlockBuffer =
          lists:sort(fun block_sorter/2,
                     [{BlockSeq, BlockValue} | BlockBuffer]),
      NewState0 = State#state{blocks_left=NewRemaining,
                              block_buffer=UpdBlockBuffer,
                              from=undefined},
      case BlocksLeft of
          0 ->
              NewState = NewState0,
              NextStateName = sending_remaining;
          _ ->
              {ReadRequests, UpdFreeReaders} =
                  read_blocks(BucketName, Key, UUID, FreeReaders, LastBlockRequested+1, TotalBlocks),
              NewState = NewState0#state{last_block_requested=LastBlockRequested+ReadRequests,
                                         free_readers=UpdFreeReaders},
              NextStateName = waiting_chunks
      end,
      {next_state, NextStateName, NewState}.
  
  sending_remaining(get_next_chunk, _From, #state{block_buffer=[{_, Block} | RestBlockBuffer]}=State) ->
      NewState = State#state{block_buffer=RestBlockBuffer},
      case RestBlockBuffer of
          [] ->
              {stop, normal, {done, Block}, NewState};
          _ ->
              {reply, {chunk, Block}, sending_remaining, NewState}
      end.
  
  %% @private
  