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:     {stop, normal, State};
  waiting_continue_or_stop(stop, State) ->
      {stop, normal, State};
  waiting_continue_or_stop(continue, #state{value_cache=CachedValue,
                                            manifest=Manifest,
                                            bucket=BucketName,
                                            key=Key,
                                            manifest_uuid=UUID,
                                            %% @TODO This only works with BLOCK_FETCH_CONCURRENCY
                                            %% of 1.
                                            free_readers=[ReaderPid | _]}=State) ->
      case CachedValue of
          undefined ->
              BlockSequences = riak_moss_lfs_utils:block_sequences_for_manifest(Manifest),
              case BlockSequences of
                  [] ->
                      %% No blocks = empty file
                      {next_state, waiting_chunk_request, State};
                  [_|_] ->
                      BlocksLeft = sets:from_list(BlockSequences),
  
                      %% start retrieving the first block
                      riak_moss_block_server:get_block(ReaderPid, BucketName, binary_to_list(Key), UUID, hd(BlockSequences)),
                      {next_state, waiting_chunks, State#state{blocks_left=BlocksLeft}}
              end;
          _ ->
              %% we don't actually have to start
              %% retrieving chunks, as we already
              %% have the value cached in our State
              {next_state, waiting_chunk_request, State}
      end.
  
  waiting_chunk_request(get_next_chunk, _From, #state{value_cache=CachedValue}=State) ->
      {stop, normal, CachedValue, State}.
  
  waiting_chunks(get_next_chunk, From, #state{chunk_queue=ChunkQueue, from=PreviousFrom}=State) when PreviousFrom =:= undefined ->
      case queue:is_empty(ChunkQueue) of
          true ->
              %% 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}};
          _ ->
              {{value, ToReturn}, NewQueue} = queue:out(ChunkQueue),
              {reply, ToReturn, waiting_chunks, State#state{chunk_queue=NewQueue}}
      end.
  
  waiting_chunks({chunk, _Pid, {ChunkSeq, ChunkReturnValue}}, #state{from=From,
                                                                     blocks_left=Remaining,
                                                                     manifest_uuid=UUID,
                                                                     key=Key,
                                                                     bucket=BucketName,
                                                                     free_readers=[ReaderPid | _],
                                                                     chunk_queue=ChunkQueue}=State) ->
      %% TODO:
      %% we don't deal with missing chunks
      %% at all here, so this pattern
      %% match will fail
      {ok, ChunkValue} = ChunkReturnValue,
      NewRemaining = sets:del_element(ChunkSeq, Remaining),
  
      %% this is just to be used
      %% in a guard later
      Empty = queue:is_empty(ChunkQueue),
  
      case sets:size(NewRemaining) of
          0 ->
              case From of
                  undefined ->
                      NewQueue = queue:in({done, ChunkValue}, ChunkQueue),
                      NewState = State#state{blocks_left=NewRemaining,
                                             chunk_queue=NewQueue},
                      {next_state, sending_remaining, NewState};
                  _ when Empty ->
                      gen_fsm:reply(From, {done, ChunkValue}),
                      NewState = State#state{blocks_left=NewRemaining, from=undefined},
                      {stop, normal, NewState}
              end;
          _ ->
              riak_moss_block_server:get_block(ReaderPid, BucketName, Key, UUID, hd(lists:sort(sets:to_list(NewRemaining)))),
              case From of
                  undefined ->
                      NewQueue = queue:in({chunk, ChunkValue}, ChunkQueue),
                      NewState = State#state{blocks_left=NewRemaining,
                                             chunk_queue=NewQueue},
                      {next_state, waiting_chunks, NewState};
                  _ ->
                      {ReplyChunk, NewQueue2} = case queue:is_empty(ChunkQueue) of
                          true ->
                              {{chunk, ChunkValue}, ChunkQueue};
                          _ ->
                              {{value, ChunkToReplyWith}, NewQueue} = queue:out(ChunkQueue),
                              {ChunkToReplyWith, NewQueue}
                      end,
                      gen_fsm:reply(From, ReplyChunk),
                      NewState = State#state{blocks_left=NewRemaining,
                                             chunk_queue=NewQueue2,
                                             from=undefined},
                      {next_state, waiting_chunks, NewState}
              end
      end.
  
  sending_remaining(get_next_chunk, _From, #state{chunk_queue=ChunkQueue}=State) ->
      {{value, Item}, Queue} = queue:out(ChunkQueue),
      NewState = State#state{chunk_queue=Queue},
      case queue:is_empty(Queue) of
          true ->
              {stop, normal, Item, NewState};
          _ ->
              {reply, Item, sending_remaining, NewState}
      end.
  
  %% @private
  
negative.code:                 mani_fsm_pid :: pid(),
                  bucket :: term(),
                  key :: term(),
                  metadata_cache :: term(),
                  block_buffer=[] :: [{pos_integer, term()}],
                  manifest :: term(),
                  manifest_uuid :: term(),
                  blocks_left :: list(),
                  test=false :: boolean(),
                  next_block=0 :: pos_integer(),
                  last_block_requested :: pos_integer(),
                  total_blocks :: pos_integer(),
                  free_readers :: [pid()],
                  all_reader_pids :: [pid()]}).
  
  