Merge pull request #134 from High-Performance-IO/memory-integration-speedup

Since caches are used to improve performance for small size IO
operation, when a read() operation that targets a size greater than the
size of the read cache line size is issued, the cache is being bypassed
and the read operation occurs directly from the shared memory segment.

Author: marcoSanti

src/posix/utils/cache/read_request_cache_mem.hpp

@@ -24,19 +24,20 @@ class ReadRequestCacheMEM {
         }
     }
 
-  protected:
+protected:
     [[nodiscard]] capio_off64_t read_request(const int fd, const capio_off64_t count,
-                                             const long tid) {
-        START_LOG(capio_syscall(SYS_gettid), "call(fd=%ld, count=%llu, tid=%ld)", fd, count, tid);
+                                             const long tid, bool use_cache = true) {
+        START_LOG(capio_syscall(SYS_gettid), "call(fd=%ld, count=%llu, tid=%ld, load_data=%s)", fd,
+                  count, tid, use_cache ? "true" : "false");
         char req[CAPIO_REQ_MAX_SIZE];
 
         // send as the last parameter to the server the maximum amount of data that can be read into
         // a single line of cache
 
         auto read_begin_offset = get_capio_fd_offset(fd);
 
-        sprintf(req, "%04d %ld %llu %llu %llu %s", CAPIO_REQUEST_READ_MEM, tid, read_begin_offset,
-                count, _max_line_size, get_capio_fd_path(fd).c_str());
+        sprintf(req, "%04d %ld %llu %llu %llu %d %s", CAPIO_REQUEST_READ_MEM, tid,
+                read_begin_offset, count, _max_line_size, use_cache, get_capio_fd_path(fd).c_str());
         LOG("Sending read request %s", req);
         buf_requests->write(req, CAPIO_REQ_MAX_SIZE);
         capio_off64_t stc_queue_read = bufs_response->at(tid)->read();
@@ -49,26 +50,21 @@ class ReadRequestCacheMEM {
             LOG("File is commited. Actual offset is: %ld", stc_queue_read);
         }
 
-        // TODO: this code is not needed as the read size is allways at maximum the size of the
-        // cache line
-        /*auto read_size = stc_queue_read;
-        while (read_size > 0) {
-            const capio_off64_t tmp_read_size =
-                read_size > _max_line_size ? _max_line_size : read_size;
-            stc_queue->read(_cache, tmp_read_size);
+        if (use_cache) {
+            stc_queue->read(_cache, stc_queue_read);
             _cache_offset = 0;
-            read_size -= tmp_read_size;
-        }*/
-
-        stc_queue->read(_cache, stc_queue_read);
-        _cache_offset = 0;
-
-        LOG("Completed fetch of data from server");
+            LOG("Completed fetch of data from server");
+        } else {
+            _actual_size = 0;
+            _cache_offset = 0;
+            LOG("Data has not been loaded from server, as load_data==false."
+                " Load will occur indipendently");
+        }
 
         return stc_queue_read;
     }
 
-  public:
+public:
     explicit ReadRequestCacheMEM(const long line_size = get_posix_read_cache_line_size())
         : _cache(nullptr), _tid(capio_syscall(SYS_gettid)), _fd(-1), _max_line_size(line_size),
           _actual_size(0), _cache_offset(0), _last_read_end(-1) {
@@ -85,7 +81,7 @@ class ReadRequestCacheMEM {
         if (_cache_offset != _actual_size) {
             _actual_size = _cache_offset = 0;
         }
-        committed                  = false;
+        committed = false;
         _real_file_size_commmitted = -1;
     }
 
@@ -97,7 +93,7 @@ class ReadRequestCacheMEM {
         if (_fd != fd) {
             LOG("changed fd from %d to %d: flushing", _fd, fd);
             flush();
-            _fd            = fd;
+            _fd = fd;
             _last_read_end = get_capio_fd_offset(fd);
         }
 
@@ -116,11 +112,22 @@ class ReadRequestCacheMEM {
             return 0;
         }
 
+        /*
+         * Check: if read size is greater than the capability of the cache line, bypass
+         * the cache and perform a read directly to the provided buffer
+         */
+        if (count > _max_line_size) {
+            LOG("count > _max_line_size. Bypassing cache. Performing read() directly to buffer.");
+            const auto _read_size = read_request(_fd, count, _tid, false);
+            stc_queue->read(static_cast<char *>(buffer), _read_size);
+            return _read_size;
+        }
+
         // Check if cache is empty or if all the content of the cache has been already consumed
         if (_actual_size == 0 || _actual_size == _cache_offset) {
             LOG("No data is present locally. performing request.");
             const auto size = count < _max_line_size ? count : _max_line_size;
-            _actual_size    = read_request(_fd, size, _tid);
+            _actual_size = read_request(_fd, size, _tid);
 
             // Update count for current request. If count exceeds _actual_size, resize it to not
             // exceeds the available size on posix application
@@ -132,7 +139,7 @@ class ReadRequestCacheMEM {
             LOG("The requested amount of data can be served without performing a request");
             _read(buffer, count);
             actual_read_size = count;
-            _last_read_end   = get_capio_fd_offset(_fd) + count;
+            _last_read_end = get_capio_fd_offset(_fd) + count;
             set_capio_fd_offset(fd, _last_read_end);
 
         } else {
@@ -153,7 +160,7 @@ class ReadRequestCacheMEM {
             LOG("actual_read_size incremented to: %ld", actual_read_size);
 
             // Compute the remaining amount of data to send to client
-            auto remaining_size       = count - first_copy_size;
+            auto remaining_size = count - first_copy_size;
             capio_off64_t copy_offset = first_copy_size;
 
             while (copy_offset < count && !committed) {

src/server/client-manager/handlers/read.hpp

@@ -57,13 +57,15 @@ inline void read_handler(const char *const str) {
 inline void read_mem_handler(const char *const str) {
     pid_t tid;
     capio_off64_t read_size, client_cache_line_size, read_begin_offset;
+    int use_cache;
     char path[PATH_MAX];
-    sscanf(str, "%ld %llu %llu %llu %s", &tid, &read_begin_offset, &read_size,
-           &client_cache_line_size, path);
+    sscanf(str, "%ld %llu %llu %llu %d %s", &tid, &read_begin_offset, &read_size,
+           &client_cache_line_size, &use_cache, path);
     START_LOG(gettid(),
               "call(tid=%d, read_begin_offset=%llu, read_size=%llu, client_cache_line_size=%llu, "
-              "path=%s)",
-              tid, read_begin_offset, read_size, client_cache_line_size, path);
+              "use_cache=%s, path=%s)",
+              tid, read_begin_offset, read_size, client_cache_line_size,
+              use_cache ? "true" : "false", path);
 
     if (storage_service->sizeOf(path) < read_begin_offset + read_size &&
         !file_manager->isCommitted(path)) {
@@ -73,25 +75,30 @@ inline void read_mem_handler(const char *const str) {
         return;
     }
 
-    LOG("Computing size of data to send: minimum between:");
-    LOG("client_cache_line_size: %llu", client_cache_line_size);
-    LOG("file_size:%llu - read_begin_offset=%llu = %llu", storage_service->sizeOf(path),
-        read_begin_offset, storage_service->sizeOf(path) - read_begin_offset);
-    auto size_to_send =
-        std::min({client_cache_line_size, (storage_service->sizeOf(path) - read_begin_offset)});
-
-    LOG("Need to sent to client %llu bytes, asking storage service to send data", size_to_send);
-    storage_service->reply_to_client(tid, path, read_begin_offset, size_to_send);
+    capio_off64_t size_to_send = storage_service->sizeOf(path);
+    if (use_cache) {
+        LOG("Computing size of data to send: minimum between:");
+        LOG("client_cache_line_size: %llu", client_cache_line_size);
+        LOG("file_size:%llu - read_begin_offset=%llu = %llu", size_to_send,
+            read_begin_offset, size_to_send - read_begin_offset);
+        size_to_send =
+            std::min({client_cache_line_size, (size_to_send - read_begin_offset)});
+    }
 
     LOG("Sending to posix app the offset up to which read.");
     if (file_manager->isCommitted(path) &&
         read_begin_offset + size_to_send >= storage_service->sizeOf(path)) {
         LOG("File is committed, and end of read >= than file size."
             " signaling it to posix application by setting offset MSB to 1");
-        size_to_send = 0x8000000000000000 | size_to_send;
+        LOG("Sending offset: %llu", 0x8000000000000000 |size_to_send);
+        client_manager->reply_to_client(tid, 0x8000000000000000 | size_to_send);
+    } else {
+        LOG("File is not committed. Sending offset: %llu", size_to_send);
+        client_manager->reply_to_client(tid, size_to_send);
     }
-    LOG("Sending offset: %llu", size_to_send);
-    client_manager->reply_to_client(tid, size_to_send);
+
+    LOG("Need to sent to client %llu bytes, asking storage service to send data", size_to_send);
+    storage_service->reply_to_client(tid, path, read_begin_offset, size_to_send);
 }
 
 #endif // READ_HPP

src/server/storage-service/CapioFile/CapioMemoryFile.hpp

@@ -16,7 +16,7 @@ class CapioMemoryFile : public CapioFile {
     std::map<std::size_t, std::vector<char>> memoryBlocks;
 
     // Static file sizes of file pages
-    static constexpr u_int32_t _pageSizeMB    = 4;
+    static constexpr u_int32_t _pageSizeMB = 4;
     static constexpr u_int64_t _pageSizeBytes = _pageSizeMB * 1024 * 1024;
 
     char *cross_page_buffer_view;
@@ -59,7 +59,7 @@ class CapioMemoryFile : public CapioFile {
         return block;
     }
 
-  public:
+public:
     explicit CapioMemoryFile(const std::string &filePath) : CapioFile(filePath) {
         cross_page_buffer_view = new char[_pageSizeBytes];
     }
@@ -193,66 +193,27 @@ class CapioMemoryFile : public CapioFile {
      */
     std::size_t writeToQueue(SPSCQueue &queue, std::size_t offset,
                              std::size_t length) const override {
+        START_LOG(gettid(), "call(offset=%llu, length=%llu)", offset, length);
         std::size_t bytesRead = 0;
 
-        const auto offsets                   = compute_offsets(offset, length);
-        unsigned long map_offset             = std::get<0>(offsets);
-        unsigned long mem_block_offset_begin = std::get<1>(offsets);
-        unsigned long buffer_view_size       = std::get<2>(offsets);
-
-        if (buffer_view_size != length) {
-            /*
-             * In this case, we have requested a view that spansa over different memory blocks and
-             * cannot be served in a contiguos manner. We allocate a temporary buffer that is used
-             * to overlap the two memory block region and then write the temporary buffer.
-             * This requires more memcpy but it is invoked only when the view overlaps and as such
-             * is not frequent.
-             */
+        while (bytesRead < length) {
+            const auto [map_offset,mem_block_offset_begin , buffer_view_size] =
+                compute_offsets(offset, length - bytesRead);
 
-
-            for (auto it = memoryBlocks.lower_bound(map_offset);
-                 it != memoryBlocks.end() && bytesRead < length; ++it) {
+            if (const auto it = memoryBlocks.lower_bound(map_offset); it != memoryBlocks.end()) {
                 auto &[blockOffset, block] = *it;
 
                 if (blockOffset >= offset + length) {
-                    break; // Past the requested range
+                    return bytesRead; // Past the requested range
                 }
 
-                // Copy the data to the temporary buffer
-                memcpy(cross_page_buffer_view + bytesRead, block.data() + mem_block_offset_begin,
-                       buffer_view_size);
+                // Copy the data to the buffer
+                queue.write(block.data() + mem_block_offset_begin, buffer_view_size);
 
                 bytesRead += buffer_view_size;
-
-                const auto updated_offsets =
-                    compute_offsets(offset + bytesRead, length - bytesRead);
-                map_offset             = std::get<0>(updated_offsets);
-                mem_block_offset_begin = std::get<1>(updated_offsets);
-                buffer_view_size       = std::get<2>(updated_offsets);
+                offset += buffer_view_size;
             }
-
-            // send the temporary buffer to the application
-            queue.write(cross_page_buffer_view, bytesRead);
-
-            return bytesRead;
         }
-
-        /*
-         * Here we have requested a read that spans over a single memory block.
-         */
-        if (const auto it = memoryBlocks.lower_bound(map_offset); it != memoryBlocks.end()) {
-            auto &[blockOffset, block] = *it;
-
-            if (blockOffset >= offset + length) {
-                return bytesRead; // Past the requested range
-            }
-
-            // Copy the data to the buffer
-            queue.write(block.data() + mem_block_offset_begin, buffer_view_size);
-
-            bytesRead += buffer_view_size;
-        }
-
         return bytesRead;
     }
 };

src/server/storage-service/capio_storage_service.hpp

@@ -10,7 +10,7 @@ class CapioStorageService {
     // TODO: put all of this conde on a different thread
 
     std::unordered_map<pid_t, SPSCQueue *> *_client_to_server_queue;
-    std::unordered_map<pid_t, SPSCQueue *> *_server_to_clien_queue;
+    std::unordered_map<pid_t, SPSCQueue *> *_server_to_client_queue;
     std::unordered_map<std::string, CapioFile *> *_stored_files;
 
     std::unordered_map<std::string, std::vector<std::tuple<capio_off64_t, capio_off64_t, pid_t>>>
@@ -33,7 +33,7 @@ class CapioStorageService {
         START_LOG(gettid(), "call()");
         _stored_files = new std::unordered_map<std::string, CapioFile *>;
         _client_to_server_queue = new std::unordered_map<pid_t, SPSCQueue *>;
-        _server_to_clien_queue = new std::unordered_map<pid_t, SPSCQueue *>;
+        _server_to_client_queue = new std::unordered_map<pid_t, SPSCQueue *>;
         _threads_waiting_for_memory_data =
             new std::unordered_map<std::string,
                                    std::vector<std::tuple<capio_off64_t, capio_off64_t, pid_t>>>;
@@ -45,7 +45,7 @@ class CapioStorageService {
         // TODO: dump files to FS
         delete _stored_files;
         delete _client_to_server_queue;
-        delete _server_to_clien_queue;
+        delete _server_to_client_queue;
         delete _threads_waiting_for_memory_data;
     }
 
@@ -116,7 +116,7 @@ class CapioStorageService {
         _client_to_server_queue->emplace(
             pid, new SPSCQueue("queue-" + std::to_string(pid) + +".cts", get_cache_lines(),
                                get_cache_line_size(), workflow_name, false));
-        _server_to_clien_queue->emplace(
+        _server_to_client_queue->emplace(
             pid, new SPSCQueue("queue-" + std::to_string(pid) + +".stc", get_cache_lines(),
                                get_cache_line_size(), workflow_name, false));
         LOG("Created communication queues");
@@ -134,7 +134,7 @@ class CapioStorageService {
         START_LOG(gettid(), "call(pid=%llu, file=%s, offset=%llu, size=%llu)", pid, file.c_str(),
                   offset, size);
 
-        getFile(file)->writeToQueue(*_server_to_clien_queue->at(pid), offset, size);
+        getFile(file)->writeToQueue(*_server_to_client_queue->at(pid), offset, size);
     }
 
     /**
@@ -155,7 +155,7 @@ class CapioStorageService {
 
     void remove_client(const pid_t pid) const {
         _client_to_server_queue->erase(pid);
-        _server_to_clien_queue->erase(pid);
+        _server_to_client_queue->erase(pid);
     }
 
     /**
@@ -173,7 +173,7 @@ class CapioStorageService {
             auto c_dir = get_capio_dir().string();
             memcpy(f, c_dir.c_str(), c_dir.length());
             memcpy(f + c_dir.size(), "/*", 2);
-            _server_to_clien_queue->at(pid)->write(f, PATH_MAX);
+            _server_to_client_queue->at(pid)->write(f, PATH_MAX);
             LOG("Return value=%llu", 1);
             return 1;
         }
@@ -183,7 +183,7 @@ class CapioStorageService {
             LOG("Sending file %s", file.c_str());
             char f[PATH_MAX + 1]{0};
             memcpy(f, file.c_str(), file.size());
-            _server_to_clien_queue->at(pid)->write(f, PATH_MAX);
+            _server_to_client_queue->at(pid)->write(f, PATH_MAX);
         }
 
         LOG("Return value=%llu", files_to_store_in_mem.size());