@@ -26,8 +26,8 @@ type BufferPool struct {
2626 keyValuesStartPoint uint64
2727 maxKeys uint64
2828 redundantBlocks uint16
29- kvBuffers []Buffer // this will act as a FIFO
30- indexBuffers [] Buffer // this will act as a min-slice where the buffer with the biggest left-offset is replaced
29+ kvBuffers []* Buffer // this will act as a FIFO
30+ indexBuffers map [ uint64 ] * Buffer
3131 File * os.File
3232 FilePath string
3333 FileSize uint64
@@ -94,8 +94,8 @@ func NewBufferPool(capacity *uint64, filePath string, maxKeys *uint64, redundant
9494 keyValuesStartPoint : header .KeyValuesStartPoint ,
9595 maxKeys : header .MaxKeys ,
9696 redundantBlocks : header .RedundantBlocks ,
97- kvBuffers : make ([]Buffer , 0 , kvCap ),
98- indexBuffers : make ([] Buffer , 0 , indexCap ),
97+ kvBuffers : make ([]* Buffer , 0 , kvCap ),
98+ indexBuffers : make (map [ uint64 ] * Buffer , indexCap ),
9999 File : file ,
100100 FilePath : filePath ,
101101 FileSize : fileSize ,
@@ -118,7 +118,7 @@ func (bp *BufferPool) Append(data []byte) (uint64, error) {
118118 start := len (bp .kvBuffers ) - 1
119119 for i := start ; i >= 0 ; i -- {
120120 // make sure you get the pointer
121- buf := & bp .kvBuffers [i ]
121+ buf := bp .kvBuffers [i ]
122122 if buf .CanAppend (bp .FileSize ) {
123123 // write the data to buffer
124124 addr := buf .Append (data )
@@ -157,13 +157,12 @@ func (bp *BufferPool) UpdateIndex(addr uint64, data []byte) error {
157157 return err
158158 }
159159
160- for i := 0 ; i < len (bp .indexBuffers ); i ++ {
161- buf := & bp .indexBuffers [i ]
162- if buf .Contains (addr ) {
163- err = buf .Replace (addr , data )
164- if err != nil {
165- return err
166- }
160+ blockLeftOffset := bp .getBlockLeftOffset (addr , entries .HeaderSizeInBytes )
161+ buf , ok := bp .indexBuffers [blockLeftOffset ]
162+ if ok {
163+ err = buf .Replace (addr , data )
164+ if err != nil {
165+ return err
167166 }
168167 }
169168
@@ -180,8 +179,8 @@ func (bp *BufferPool) ClearFile() error {
180179 return err
181180 }
182181 bp .FileSize = uint64 (fileSize )
183- bp .indexBuffers = make ([] Buffer , 0 , bp .indexCapacity )
184- bp .kvBuffers = make ([] Buffer , 0 , bp .kvCapacity )
182+ bp .indexBuffers = make (map [ uint64 ] * Buffer , bp .indexCapacity )
183+ bp .kvBuffers = bp .kvBuffers [: 0 ]
185184 return nil
186185}
187186
@@ -272,8 +271,8 @@ func (bp *BufferPool) CompactFile() error {
272271 }
273272
274273 // clean up the buffers and update metadata
275- bp .kvBuffers = make ([] Buffer , 0 , bp .kvCapacity )
276- bp .indexBuffers = make ([] Buffer , 0 , bp .indexCapacity )
274+ bp .kvBuffers = bp .kvBuffers [: 0 ]
275+ bp .indexBuffers = make (map [ uint64 ] * Buffer , bp .indexCapacity )
277276 bp .File = newFile
278277 bp .FileSize = uint64 (newFileOffset )
279278
@@ -298,7 +297,7 @@ func (bp *BufferPool) GetValue(kvAddress uint64, key []byte) (*Value, error) {
298297 // since the latest kv_buffers are the ones updated when new changes occur
299298 kvBufLen := len (bp .kvBuffers )
300299 for i := kvBufLen - 1 ; i >= 0 ; i -- {
301- buf := & bp .kvBuffers [i ]
300+ buf := bp .kvBuffers [i ]
302301 if buf .Contains (kvAddress ) {
303302 return buf .GetValue (kvAddress , key )
304303 }
@@ -316,7 +315,7 @@ func (bp *BufferPool) GetValue(kvAddress uint64, key []byte) (*Value, error) {
316315 }
317316
318317 // update kv_buffers only upto actual data read (cater for partially filled buffer)
319- bp .kvBuffers = append (bp .kvBuffers , * NewBuffer (kvAddress , buf [:bytesRead ], bp .bufferSize ))
318+ bp .kvBuffers = append (bp .kvBuffers , NewBuffer (kvAddress , buf [:bytesRead ], bp .bufferSize ))
320319 entry , err := entries .ExtractKeyValueEntryFromByteArray (buf , 0 )
321320 if err != nil {
322321 return nil , err
@@ -338,7 +337,7 @@ func (bp *BufferPool) TryDeleteKvEntry(kvAddress uint64, key []byte) (bool, erro
338337 // since the latest kv_buffers are the ones updated when new changes occur
339338 kvBufLen := len (bp .kvBuffers )
340339 for i := kvBufLen - 1 ; i >= 0 ; i -- {
341- buf := & bp .kvBuffers [i ]
340+ buf := bp .kvBuffers [i ]
342341 if buf .Contains (kvAddress ) {
343342 success , err := buf .TryDeleteKvEntry (kvAddress , key )
344343 if err != nil {
@@ -367,6 +366,8 @@ func (bp *BufferPool) TryDeleteKvEntry(kvAddress uint64, key []byte) (bool, erro
367366 if err != nil {
368367 return false , err
369368 }
369+
370+ return true , nil
370371 }
371372
372373 return false , nil
@@ -386,7 +387,7 @@ func (bp *BufferPool) AddrBelongsToKey(kvAddress uint64, key []byte) (bool, erro
386387 // since the latest kv_buffers are the ones updated when new changes occur
387388 kvBufLen := len (bp .kvBuffers )
388389 for i := kvBufLen - 1 ; i >= 0 ; i -- {
389- buf := & bp .kvBuffers [i ]
390+ buf := bp .kvBuffers [i ]
390391 if buf .Contains (kvAddress ) {
391392 return buf .AddrBelongsToKey (kvAddress , key )
392393 }
@@ -404,7 +405,7 @@ func (bp *BufferPool) AddrBelongsToKey(kvAddress uint64, key []byte) (bool, erro
404405 }
405406
406407 // update kv_buffers only upto actual data read (cater for partially filled buffer)
407- bp .kvBuffers = append (bp .kvBuffers , * NewBuffer (kvAddress , buf [:bytesRead ], bp .bufferSize ))
408+ bp .kvBuffers = append (bp .kvBuffers , NewBuffer (kvAddress , buf [:bytesRead ], bp .bufferSize ))
408409
409410 keyInFile := buf [entries .OffsetForKeyInKVArray : entries .OffsetForKeyInKVArray + uint64 (len (key ))]
410411 isForKey := bytes .Contains (keyInFile , key )
@@ -421,48 +422,47 @@ func (bp *BufferPool) ReadIndex(addr uint64) ([]byte, error) {
421422 return nil , err
422423 }
423424
424- size := entries .IndexEntrySizeInBytes
425- // starts from buffer with lowest left_offset, which I expect to have more keys
426- idxBufLen := len (bp .indexBuffers )
427- for i := 0 ; i < idxBufLen ; i ++ {
428- buf := & bp .indexBuffers [i ]
429- if buf .Contains (addr ) {
430- return buf .ReadAt (addr , size )
431- }
425+ blockLeftOffset := bp .getBlockLeftOffset (addr , entries .HeaderSizeInBytes )
426+ buf , ok := bp .indexBuffers [blockLeftOffset ]
427+ if ok {
428+ return buf .ReadAt (addr , entries .IndexEntrySizeInBytes )
432429 }
433430
434- buf := make ([]byte , bp .bufferSize )
435- bytesRead , err := bp .File .ReadAt (buf , int64 (addr ))
431+ data := make ([]byte , bp .bufferSize )
432+ // Index buffers should have preset boundaries matching
433+ // StartOfIndex - StartOfIndex + BlockSize,
434+ // StartOfIndex + BlockSize - StartOfIndex + (2*BlockSize)
435+ // StartOfIndex + (2*BlockSize) - StartOfIndex + (3*BlockSize) ...
436+ _ , err = bp .File .ReadAt (data , int64 (blockLeftOffset ))
436437 if err != nil && ! errors .Is (err , io .EOF ) {
437438 return nil , err
438439 }
439440
440- // update kv_buffers only upto actual data read (cater for partially filled buffer)
441- newIdxBuf := NewBuffer (addr , buf [:bytesRead ], bp .bufferSize )
442-
443- if uint64 (idxBufLen ) >= bp .indexCapacity {
444- // we wish to remove the last buf, and replace it with the new one
445- // but maintain the ascending order of LeftOffsets
446- // we will start at the second-last buf and move towards the front of the slice
447- for i := idxBufLen - 2 ; i >= 0 ; i -- {
448- currOffset := (& bp .indexBuffers [i ]).LeftOffset
449- if currOffset < newIdxBuf .LeftOffset || i == 0 {
450- // copy the new buffer in previous position and stop if
451- // the current buffer has a lower left offset or if we
452- // have reached the end of the array
453- bp .indexBuffers [i + 1 ] = * newIdxBuf
454- break
455- } else {
456- // move current buf backwards
457- bp .indexBuffers [i + 1 ] = bp .indexBuffers [i ]
441+ if uint64 (len (bp .indexBuffers )) >= bp .indexCapacity {
442+ biggestLeftOffset := uint64 (0 )
443+ for lftOffset , _ := range bp .indexBuffers {
444+ if lftOffset >= biggestLeftOffset {
445+ biggestLeftOffset = lftOffset
458446 }
459447 }
448+
449+ // delete the buffer with the biggest left offset as those with lower left offsets
450+ // are expected to have more keys
451+ delete (bp .indexBuffers , biggestLeftOffset )
452+ bp .indexBuffers [blockLeftOffset ] = NewBuffer (blockLeftOffset , data , bp .bufferSize )
460453 } else {
461- // Just append
462- bp .indexBuffers = append (bp .indexBuffers , * newIdxBuf )
454+ bp .indexBuffers [blockLeftOffset ] = NewBuffer (blockLeftOffset , data , bp .bufferSize )
463455 }
464456
465- return buf [:size ], nil
457+ start := addr - blockLeftOffset
458+ return data [start : start + entries .IndexEntrySizeInBytes ], nil
459+ }
460+
461+ // getBlockLeftOffset returns the left offset for the block in which the address is to be found
462+ func (bp * BufferPool ) getBlockLeftOffset (addr uint64 , minOffset uint64 ) uint64 {
463+ blockPosition := (addr - minOffset ) / bp .bufferSize
464+ blockLeftOffset := (blockPosition * bp .bufferSize ) + minOffset
465+ return blockLeftOffset
466466}
467467
468468// Eq checks that other is equal to bp
@@ -482,13 +482,13 @@ func (bp *BufferPool) Eq(other *BufferPool) bool {
482482 }
483483
484484 for i , buf := range bp .kvBuffers {
485- if ! buf .Eq (& other .kvBuffers [i ]) {
485+ if ! buf .Eq (other .kvBuffers [i ]) {
486486 return false
487487 }
488488 }
489489
490490 for i , buf := range bp .indexBuffers {
491- if ! buf .Eq (& other .indexBuffers [i ]) {
491+ if ! buf .Eq (other .indexBuffers [i ]) {
492492 return false
493493 }
494494 }
0 commit comments