buffer_pool.h

Go to the documentation of this file.

/* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * *
 * Distributed under BSD 3-Clause license.                                   *
 * Copyright by The HDF Group.                                               *
 * Copyright by the Illinois Institute of Technology.                        *
 * All rights reserved.                                                      *
 *                                                                           *
 * This file is part of Hermes. The full Hermes copyright notice, including  *
 * terms governing use, modification, and redistribution, is contained in    *
 * the COPYING file, which can be found at the top directory. If you do not  *
 * have access to the file, you may request a copy from help@hdfgroup.org.   *
 * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */
 
#ifndef HERMES_BUFFER_POOL_H_
#define HERMES_BUFFER_POOL_H_
 
#include <assert.h>
#include <stddef.h>
#include <stdio.h>
 
#include <atomic>
#include <string>
#include <utility>
#include <vector>
 
#include "communication.h"
#include "hermes_status.h"
#include "hermes_types.h"
#include "memory_management.h"
 
namespace hermes {
 
struct RpcContext;
 
struct Device {
  f32 bandwidth_mbps;
  f32 latency_ns;
  DeviceID id;
  bool is_byte_addressable;
  bool has_fallocate;
  bool is_shared;
  char mount_point[kMaxPathLength];
};
 
struct Target {
  TargetID id; 
  u64 capacity;
  std::atomic<u64> remaining_space; 
  std::atomic<u64> speed;           
  ChunkedIdList effective_blobs;    
  TicketMutex effective_blobs_lock; 
};
 
union BufferID {
  struct {
    u32 header_index;
    u32 node_id;
  } bits;
 
  u64 as_int;
};
 
struct BufferIdHash {
  size_t operator()(const BufferID &id) const {
    return std::hash<u64>()(id.as_int);
  }
};
 
bool operator==(const BufferID &lhs, const BufferID &rhs);
 
struct ShmemClientInfo {
  ptrdiff_t mdm_offset; 
  ptrdiff_t bpm_offset; 
};
 
struct BufferHeader {
  BufferID id;
  BufferID next_free;
  ptrdiff_t data_offset;
  u32 used;
  u32 capacity;
  DeviceID device_id;
  bool in_use;
  std::atomic<bool> locked;
};
 
struct BufferPool {
  ptrdiff_t headers_offset;
  ptrdiff_t devices_offset;
  ptrdiff_t targets_offset;
  ptrdiff_t free_list_offsets[kMaxDevices];
  ptrdiff_t slab_unit_sizes_offsets[kMaxDevices];
  ptrdiff_t slab_buffer_sizes_offsets[kMaxDevices];
  ptrdiff_t buffers_available_offsets[kMaxDevices];
  TicketMutex ticket_mutex;
 
  std::atomic<i64> capacity_adjustments[kMaxDevices];
 
  i32 block_sizes[kMaxDevices];
  i32 num_slabs[kMaxDevices];
  u32 num_headers[kMaxDevices];
  i32 num_devices;
  i32 num_targets;
  u32 total_headers;
  f32 min_device_bw_mbps;
  f32 max_device_bw_mbps;
};
 
struct BufferOrganizer;
 
struct SharedMemoryContext {
  u8 *shm_base;
  ptrdiff_t buffer_pool_offset;
  ptrdiff_t metadata_manager_offset;
  u64 shm_size;
  BufferOrganizer *bo;
 
  // File buffering context
  std::vector<std::vector<std::string>> buffering_filenames;
  int open_files[kMaxDevices][kMaxBufferPoolSlabs];
  FILE *swap_file;
};
 
struct BufferIdArray;
 
size_t GetBlobSize(SharedMemoryContext *context, RpcContext *rpc,
                   BufferIdArray *buffer_ids);
 
size_t GetBlobSizeById(SharedMemoryContext *context, RpcContext *rpc,
                       Arena *arena, BlobID blob_id);
 
f32 GetBlobImportanceScore(SharedMemoryContext *context, RpcContext *rpc,
                           BlobID blob_id);
void MakeFullShmemName(char *dest, const char *base);
 
void InitFilesForBuffering(SharedMemoryContext *context,
                           CommunicationContext &comm);
 
SharedMemoryContext GetSharedMemoryContext(char *shmem_name);
 
void ReleaseSharedMemoryContext(SharedMemoryContext *context);
 
void UnmapSharedMemory(SharedMemoryContext *context);
 
std::vector<BufferID> GetBuffers(SharedMemoryContext *context,
                                 const PlacementSchema &schema);
void ReleaseBuffers(SharedMemoryContext *context, RpcContext *rpc,
                    const std::vector<BufferID> &buffer_ids);
void StartBufferPoolRpcServer(SharedMemoryContext *context, const char *addr,
                              i32 num_rpc_threads);
 
void Finalize(SharedMemoryContext *context, CommunicationContext *comm,
              RpcContext *rpc, const char *shmem_name, Arena *trans_arena,
              bool is_application_core, bool force_rpc_shutdown);
 
// I/O Clients
 
struct Blob {
  u8 *data;
  u64 size;
};
 
struct SwapBlob {
  u32 node_id;        
  u64 offset;         
  u64 size;           
  BucketID bucket_id; 
};
 
enum SwapBlobMembers {
  SwapBlobMembers_NodeId,
  SwapBlobMembers_Offset,
  SwapBlobMembers_Size,
  SwapBlobMembers_BucketId,
 
  SwapBlobMembers_Count
};
 
void WriteBlobToBuffers(SharedMemoryContext *context, RpcContext *rpc,
                        const Blob &blob,
                        const std::vector<BufferID> &buffer_ids);
 
size_t ReadBlobFromBuffers(SharedMemoryContext *context, RpcContext *rpc,
                           Blob *blob, BufferIdArray *buffer_ids,
                           u32 *buffer_sizes);
 
size_t ReadBlobById(SharedMemoryContext *context, RpcContext *rpc, Arena *arena,
                    Blob blob, BlobID blob_id);
 
size_t ReadBlobById(SharedMemoryContext *context, RpcContext *rpc, Arena *arena,
                    api::Blob &dest, BlobID blob_id);
 
size_t LocalWriteBufferById(SharedMemoryContext *context, BufferID id,
                            const Blob &blob, size_t offset);
size_t LocalReadBufferById(SharedMemoryContext *context, BufferID id,
                           Blob *blob, size_t offset);
 
SwapBlob PutToSwap(SharedMemoryContext *context, RpcContext *rpc,
                   const std::string &name, BucketID bucket_id, const u8 *data,
                   size_t size);
template <typename T>
std::vector<SwapBlob> PutToSwap(SharedMemoryContext *context, RpcContext *rpc,
                                BucketID id, std::vector<std::vector<T>> &blobs,
                                std::vector<std::string> &names) {
  size_t num_blobs = blobs.size();
  std::vector<SwapBlob> result(num_blobs);
 
  for (size_t i = 0; i < num_blobs; ++i) {
    SwapBlob swap_blob =
        PutToSwap(context, rpc, names[i], id, (const u8 *)blobs[i].data(),
                  blobs[i].size() * sizeof(T));
    result.push_back(swap_blob);
  }
 
  return result;
}
 
SwapBlob WriteToSwap(SharedMemoryContext *context, Blob blob, BlobID blob_id,
                     BucketID bucket_id);
size_t ReadFromSwap(SharedMemoryContext *context, Blob blob,
                    SwapBlob swap_blob);
 
std::vector<f32> GetBandwidths(SharedMemoryContext *context,
                               const std::vector<TargetID> &targets);
 
u32 GetNumBuffersAvailable(SharedMemoryContext *context, DeviceID device_id);
u32 GetBufferSize(SharedMemoryContext *context, RpcContext *rpc, BufferID id);
bool BufferIsByteAddressable(SharedMemoryContext *context, BufferID id);
api::Status PlaceInHierarchy(SharedMemoryContext *context, RpcContext *rpc,
                             SwapBlob swap_blob, const std::string &blob_name,
                             const api::Context &ctx);
api::Status PlaceBlob(SharedMemoryContext *context, RpcContext *rpc,
                      PlacementSchema &schema, Blob blob,
                      const std::string &name, BucketID bucket_id,
                      const api::Context &ctx,
                      bool called_from_buffer_organizer = false);
api::Status StdIoPersistBucket(SharedMemoryContext *context, RpcContext *rpc,
                               Arena *arena, BucketID bucket_id,
                               const std::string &file_name,
                               const std::string &open_mode);
 
api::Status StdIoPersistBlob(SharedMemoryContext *context, RpcContext *rpc,
                             Arena *arena, BlobID blob_id, int fd,
                             const i32 &offset);
 
Device *GetDeviceFromHeader(SharedMemoryContext *context, BufferHeader *header);
}  // namespace hermes
 
#endif  // HERMES_BUFFER_POOL_H_