From d8b975f4e73ae940c0c0f9c8c6c7aac1199fee09 Mon Sep 17 00:00:00 2001 From: Martin Sustrik Date: Sat, 28 Aug 2010 13:14:45 +0200 Subject: msg_store_t renamed to swap_t --- src/Makefile.am | 4 +- src/msg_store.cpp | 307 ------------------------------------------------------ src/msg_store.hpp | 114 -------------------- src/pipe.cpp | 2 +- src/pipe.hpp | 4 +- src/swap.cpp | 307 ++++++++++++++++++++++++++++++++++++++++++++++++++++++ src/swap.hpp | 114 ++++++++++++++++++++ 7 files changed, 426 insertions(+), 426 deletions(-) delete mode 100644 src/msg_store.cpp delete mode 100644 src/msg_store.hpp create mode 100644 src/swap.cpp create mode 100644 src/swap.hpp (limited to 'src') diff --git a/src/Makefile.am b/src/Makefile.am index d7509dc..f15814f 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -76,7 +76,6 @@ libzmq_la_SOURCES = \ lb.hpp \ likely.hpp \ msg_content.hpp \ - msg_store.hpp \ mutex.hpp \ named_session.hpp \ object.hpp \ @@ -105,6 +104,7 @@ libzmq_la_SOURCES = \ stdint.hpp \ streamer.hpp \ sub.hpp \ + swap.hpp \ tcp_connecter.hpp \ tcp_listener.hpp \ tcp_socket.hpp \ @@ -138,7 +138,6 @@ libzmq_la_SOURCES = \ ip.cpp \ kqueue.cpp \ lb.cpp \ - msg_store.cpp \ named_session.cpp \ object.cpp \ options.cpp \ @@ -162,6 +161,7 @@ libzmq_la_SOURCES = \ socket_base.cpp \ streamer.cpp \ sub.cpp \ + swap.cpp \ tcp_connecter.cpp \ tcp_listener.cpp \ tcp_socket.cpp \ diff --git a/src/msg_store.cpp b/src/msg_store.cpp deleted file mode 100644 index aaf6dbe..0000000 --- a/src/msg_store.cpp +++ /dev/null @@ -1,307 +0,0 @@ -/* - Copyright (c) 2007-2010 iMatix Corporation - - This file is part of 0MQ. - - 0MQ is free software; you can redistribute it and/or modify it under - the terms of the Lesser GNU General Public License as published by - the Free Software Foundation; either version 3 of the License, or - (at your option) any later version. - - 0MQ is distributed in the hope that it will be useful, - but WITHOUT ANY WARRANTY; without even the implied warranty of - MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - Lesser GNU General Public License for more details. - - You should have received a copy of the Lesser GNU General Public License - along with this program. If not, see . -*/ - -#include "platform.hpp" - -#ifdef ZMQ_HAVE_WINDOWS -#include "windows.hpp" -#include -#else -#include -#endif - -#include "../include/zmq.h" - -#include -#include -#include -#include -#include -#include - -#include "atomic_counter.hpp" -#include "msg_store.hpp" -#include "err.hpp" - -zmq::msg_store_t::msg_store_t (int64_t filesize_, size_t block_size_) : - fd (-1), - filesize (filesize_), - file_pos (0), - write_pos (0), - read_pos (0), - block_size (block_size_), - write_buf_start_addr (0) -{ - zmq_assert (filesize > 0); - zmq_assert (block_size > 0); - - buf1 = new (std::nothrow) char [block_size]; - zmq_assert (buf1); - - buf2 = new (std::nothrow) char [block_size]; - zmq_assert (buf2); - - read_buf = write_buf = buf1; -} - -zmq::msg_store_t::~msg_store_t () -{ - delete [] buf1; - delete [] buf2; - - if (fd == -1) - return; - -#ifdef ZMQ_HAVE_WINDOWS - int rc = _close (fd); -#else - int rc = close (fd); -#endif - errno_assert (rc == 0); - -#ifdef ZMQ_HAVE_WINDOWS - rc = _unlink (filename.c_str ()); -#else - rc = unlink (filename.c_str ()); -#endif - errno_assert (rc == 0); -} - -int zmq::msg_store_t::init () -{ - static zmq::atomic_counter_t seqnum (0); - - // Get process ID. -#ifdef ZMQ_HAVE_WINDOWS - int pid = GetCurrentThreadId (); -#else - pid_t pid = getpid (); -#endif - - std::ostringstream outs; - outs << "zmq_" << pid << '_' << seqnum.get () << ".swap"; - filename = outs.str (); - - seqnum.add (1); - - // Open the backing file. -#ifdef ZMQ_HAVE_WINDOWS - fd = _open (filename.c_str (), _O_RDWR | _O_CREAT, 0600); -#else - fd = open (filename.c_str (), O_RDWR | O_CREAT, 0600); -#endif - if (fd == -1) - return -1; - -#ifdef ZMQ_HAVE_LINUX - // Enable more aggresive read-ahead optimization. - posix_fadvise (fd, 0, filesize, POSIX_FADV_SEQUENTIAL); -#endif - return 0; -} - -bool zmq::msg_store_t::store (zmq_msg_t *msg_) -{ - size_t msg_size = zmq_msg_size (msg_); - - // Check buffer space availability. - // NOTE: We always keep one byte open. - if (buffer_space () <= (int64_t) (sizeof msg_size + 1 + msg_size)) - return false; - - // Don't store the ZMQ_MSG_SHARED flag. - uint8_t msg_flags = msg_->flags & ~ZMQ_MSG_SHARED; - - // Write message length, flags, and message body. - copy_to_file (&msg_size, sizeof msg_size); - copy_to_file (&msg_flags, sizeof msg_flags); - copy_to_file (zmq_msg_data (msg_), msg_size); - - zmq_msg_close (msg_); - - return true; -} - -void zmq::msg_store_t::fetch (zmq_msg_t *msg_) -{ - // There must be at least one message available. - zmq_assert (read_pos != write_pos); - - // Retrieve the message size. - size_t msg_size; - copy_from_file (&msg_size, sizeof msg_size); - - // Initialize the message. - zmq_msg_init_size (msg_, msg_size); - - // Retrieve the message flags. - copy_from_file (&msg_->flags, sizeof msg_->flags); - - // Retrieve the message payload. - copy_from_file (zmq_msg_data (msg_), msg_size); -} - -void zmq::msg_store_t::commit () -{ - commit_pos = write_pos; -} - -void zmq::msg_store_t::rollback () -{ - if (commit_pos == write_pos || read_pos == write_pos) - return; - - if (write_pos > read_pos) - zmq_assert (read_pos <= commit_pos && commit_pos <= write_pos); - else - zmq_assert (read_pos <= commit_pos || commit_pos <= write_pos); - - if (commit_pos / block_size == read_pos / block_size) { - write_buf_start_addr = commit_pos % block_size; - write_buf = read_buf; - } - else if (commit_pos / block_size != write_pos / block_size) { - write_buf_start_addr = commit_pos % block_size; - fill_buf (write_buf, write_buf_start_addr); - } - write_pos = commit_pos; -} - -bool zmq::msg_store_t::empty () -{ - return read_pos == write_pos; -} - -bool zmq::msg_store_t::full () -{ - return buffer_space () == 1; -} - -void zmq::msg_store_t::copy_from_file (void *buffer_, size_t count_) -{ - char *dest_ptr = (char *) buffer_; - size_t chunk_size, remainder = count_; - - while (remainder > 0) { - chunk_size = std::min (remainder, - std::min ((size_t) (filesize - read_pos), - (size_t) (block_size - read_pos % block_size))); - - memcpy (dest_ptr, &read_buf [read_pos % block_size], chunk_size); - dest_ptr += chunk_size; - - read_pos = (read_pos + chunk_size) % filesize; - if (read_pos % block_size == 0) { - if (read_pos / block_size == write_pos / block_size) - read_buf = write_buf; - else - fill_buf (read_buf, read_pos); - } - remainder -= chunk_size; - } -} - -void zmq::msg_store_t::copy_to_file (const void *buffer_, size_t count_) -{ - char *source_ptr = (char *) buffer_; - size_t chunk_size, remainder = count_; - - while (remainder > 0) { - chunk_size = std::min (remainder, - std::min ((size_t) (filesize - write_pos), - (size_t) (block_size - write_pos % block_size))); - - memcpy (&write_buf [write_pos % block_size], source_ptr, chunk_size); - source_ptr += chunk_size; - - write_pos = (write_pos + chunk_size) % filesize; - if (write_pos % block_size == 0) { - save_write_buf (); - write_buf_start_addr = write_pos; - - if (write_buf == read_buf) { - if (read_buf == buf2) - write_buf = buf1; - else - write_buf = buf2; - } - } - remainder -= chunk_size; - } -} - -void zmq::msg_store_t::fill_buf (char *buf, int64_t pos) -{ - if (file_pos != pos) { -#ifdef ZMQ_HAVE_WINDOWS - __int64 offset = _lseeki64 (fd, pos, SEEK_SET); -#else - off_t offset = lseek (fd, (off_t) pos, SEEK_SET); -#endif - errno_assert (offset == pos); - file_pos = pos; - } - size_t octets_stored = 0; - size_t octets_total = std::min (block_size, (size_t) (filesize - file_pos)); - - while (octets_stored < octets_total) { -#ifdef ZMQ_HAVE_WINDOWS - int rc = _read (fd, &buf [octets_stored], octets_total - octets_stored); -#else - ssize_t rc = read (fd, &buf [octets_stored], octets_total - octets_stored); -#endif - errno_assert (rc > 0); - octets_stored += rc; - } - file_pos += octets_total; -} - -void zmq::msg_store_t::save_write_buf () -{ - if (file_pos != write_buf_start_addr) { -#ifdef ZMQ_HAVE_WINDOWS - __int64 offset = _lseeki64 (fd, write_buf_start_addr, SEEK_SET); -#else - off_t offset = lseek (fd, (off_t) write_buf_start_addr, SEEK_SET); -#endif - errno_assert (offset == write_buf_start_addr); - file_pos = write_buf_start_addr; - } - size_t octets_stored = 0; - size_t octets_total = std::min (block_size, (size_t) (filesize - file_pos)); - - while (octets_stored < octets_total) { -#ifdef ZMQ_HAVE_WINDOWS - int rc = _write (fd, &write_buf [octets_stored], octets_total - octets_stored); -#else - ssize_t rc = write (fd, &write_buf [octets_stored], octets_total - octets_stored); -#endif - errno_assert (rc > 0); - octets_stored += rc; - } - file_pos += octets_total; -} - -int64_t zmq::msg_store_t::buffer_space () -{ - if (write_pos < read_pos) - return read_pos - write_pos; - - return filesize - (write_pos - read_pos); -} diff --git a/src/msg_store.hpp b/src/msg_store.hpp deleted file mode 100644 index 765fc60..0000000 --- a/src/msg_store.hpp +++ /dev/null @@ -1,114 +0,0 @@ -/* - Copyright (c) 2007-2010 iMatix Corporation - - This file is part of 0MQ. - - 0MQ is free software; you can redistribute it and/or modify it under - the terms of the Lesser GNU General Public License as published by - the Free Software Foundation; either version 3 of the License, or - (at your option) any later version. - - 0MQ is distributed in the hope that it will be useful, - but WITHOUT ANY WARRANTY; without even the implied warranty of - MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - Lesser GNU General Public License for more details. - - You should have received a copy of the Lesser GNU General Public License - along with this program. If not, see . -*/ - -#ifndef __ZMQ_MSG_STORE_HPP_INCLUDED__ -#define __ZMQ_MSG_STORE_HPP_INCLUDED__ - -#include "../include/zmq.h" - -#include -#include "stdint.hpp" - -namespace zmq -{ - - // This class implements a message store. Messages are retrieved from - // the store in the same order as they entered it. - - class msg_store_t - { - public: - - enum { default_block_size = 8192 }; - - // Creates message store. - msg_store_t (int64_t filesize_, size_t block_size_ = default_block_size); - - ~msg_store_t (); - - int init (); - - // Stores the message into the message store. The function - // returns false if the message store is full; true otherwise. - bool store (zmq_msg_t *msg_); - - // Fetches the oldest message from the message store. It is an error - // to call this function when the message store is empty. - void fetch (zmq_msg_t *msg_); - - void commit (); - - void rollback (); - - // Returns true if the message store is empty; false otherwise. - bool empty (); - - // Returns true if and only if the store is full. - bool full (); - - private: - - // Copies data from a memory buffer to the backing file. - // Wraps around when reaching maximum file size. - void copy_from_file (void *buffer_, size_t count_); - - // Copies data from the backing file to the memory buffer. - // Wraps around when reaching end-of-file. - void copy_to_file (const void *buffer_, size_t count_); - - // Returns the buffer space available. - int64_t buffer_space (); - - void fill_buf (char *buf, int64_t pos); - - void save_write_buf (); - - // File descriptor to the backing file. - int fd; - - // Name of the backing file. - std::string filename; - - // Maximum size of the backing file. - int64_t filesize; - - // File offset associated with the fd file descriptor. - int64_t file_pos; - - // File offset the next message will be stored at. - int64_t write_pos; - - // File offset the next message will be read from. - int64_t read_pos; - - int64_t commit_pos; - - size_t block_size; - - char *buf1; - char *buf2; - char *read_buf; - char *write_buf; - - int64_t write_buf_start_addr; - }; - -} - -#endif diff --git a/src/pipe.cpp b/src/pipe.cpp index de20392..0108d56 100644 --- a/src/pipe.cpp +++ b/src/pipe.cpp @@ -174,7 +174,7 @@ zmq::writer_t::writer_t (object_t *parent_, pipe_t *pipe_, reader_t *reader_, // Open the swap file, if required. if (swap_size_ > 0) { - swap = new (std::nothrow) msg_store_t (swap_size_); + swap = new (std::nothrow) swap_t (swap_size_); zmq_assert (swap); int rc = swap->init (); zmq_assert (rc == 0); diff --git a/src/pipe.hpp b/src/pipe.hpp index dee37a5..d85f9dc 100644 --- a/src/pipe.hpp +++ b/src/pipe.hpp @@ -25,7 +25,7 @@ #include "stdint.hpp" #include "yarray_item.hpp" #include "ypipe.hpp" -#include "msg_store.hpp" +#include "swap.hpp" #include "config.hpp" #include "object.hpp" @@ -183,7 +183,7 @@ namespace zmq // Pointer to the message swap. If NULL, messages are always // kept in main memory. - msg_store_t *swap; + swap_t *swap; // Sink for the events (either the socket or the session). i_writer_events *sink; diff --git a/src/swap.cpp b/src/swap.cpp new file mode 100644 index 0000000..cf73576 --- /dev/null +++ b/src/swap.cpp @@ -0,0 +1,307 @@ +/* + Copyright (c) 2007-2010 iMatix Corporation + + This file is part of 0MQ. + + 0MQ is free software; you can redistribute it and/or modify it under + the terms of the Lesser GNU General Public License as published by + the Free Software Foundation; either version 3 of the License, or + (at your option) any later version. + + 0MQ is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + Lesser GNU General Public License for more details. + + You should have received a copy of the Lesser GNU General Public License + along with this program. If not, see . +*/ + +#include "platform.hpp" + +#ifdef ZMQ_HAVE_WINDOWS +#include "windows.hpp" +#include +#else +#include +#endif + +#include "../include/zmq.h" + +#include +#include +#include +#include +#include +#include + +#include "swap.hpp" +#include "atomic_counter.hpp" +#include "err.hpp" + +zmq::swap_t::swap_t (int64_t filesize_, size_t block_size_) : + fd (-1), + filesize (filesize_), + file_pos (0), + write_pos (0), + read_pos (0), + block_size (block_size_), + write_buf_start_addr (0) +{ + zmq_assert (filesize > 0); + zmq_assert (block_size > 0); + + buf1 = new (std::nothrow) char [block_size]; + zmq_assert (buf1); + + buf2 = new (std::nothrow) char [block_size]; + zmq_assert (buf2); + + read_buf = write_buf = buf1; +} + +zmq::swap_t::~swap_t () +{ + delete [] buf1; + delete [] buf2; + + if (fd == -1) + return; + +#ifdef ZMQ_HAVE_WINDOWS + int rc = _close (fd); +#else + int rc = close (fd); +#endif + errno_assert (rc == 0); + +#ifdef ZMQ_HAVE_WINDOWS + rc = _unlink (filename.c_str ()); +#else + rc = unlink (filename.c_str ()); +#endif + errno_assert (rc == 0); +} + +int zmq::swap_t::init () +{ + static zmq::atomic_counter_t seqnum (0); + + // Get process ID. +#ifdef ZMQ_HAVE_WINDOWS + int pid = GetCurrentThreadId (); +#else + pid_t pid = getpid (); +#endif + + std::ostringstream outs; + outs << "zmq_" << pid << '_' << seqnum.get () << ".swap"; + filename = outs.str (); + + seqnum.add (1); + + // Open the backing file. +#ifdef ZMQ_HAVE_WINDOWS + fd = _open (filename.c_str (), _O_RDWR | _O_CREAT, 0600); +#else + fd = open (filename.c_str (), O_RDWR | O_CREAT, 0600); +#endif + if (fd == -1) + return -1; + +#ifdef ZMQ_HAVE_LINUX + // Enable more aggresive read-ahead optimization. + posix_fadvise (fd, 0, filesize, POSIX_FADV_SEQUENTIAL); +#endif + return 0; +} + +bool zmq::swap_t::store (zmq_msg_t *msg_) +{ + size_t msg_size = zmq_msg_size (msg_); + + // Check buffer space availability. + // NOTE: We always keep one byte open. + if (buffer_space () <= (int64_t) (sizeof msg_size + 1 + msg_size)) + return false; + + // Don't store the ZMQ_MSG_SHARED flag. + uint8_t msg_flags = msg_->flags & ~ZMQ_MSG_SHARED; + + // Write message length, flags, and message body. + copy_to_file (&msg_size, sizeof msg_size); + copy_to_file (&msg_flags, sizeof msg_flags); + copy_to_file (zmq_msg_data (msg_), msg_size); + + zmq_msg_close (msg_); + + return true; +} + +void zmq::swap_t::fetch (zmq_msg_t *msg_) +{ + // There must be at least one message available. + zmq_assert (read_pos != write_pos); + + // Retrieve the message size. + size_t msg_size; + copy_from_file (&msg_size, sizeof msg_size); + + // Initialize the message. + zmq_msg_init_size (msg_, msg_size); + + // Retrieve the message flags. + copy_from_file (&msg_->flags, sizeof msg_->flags); + + // Retrieve the message payload. + copy_from_file (zmq_msg_data (msg_), msg_size); +} + +void zmq::swap_t::commit () +{ + commit_pos = write_pos; +} + +void zmq::swap_t::rollback () +{ + if (commit_pos == write_pos || read_pos == write_pos) + return; + + if (write_pos > read_pos) + zmq_assert (read_pos <= commit_pos && commit_pos <= write_pos); + else + zmq_assert (read_pos <= commit_pos || commit_pos <= write_pos); + + if (commit_pos / block_size == read_pos / block_size) { + write_buf_start_addr = commit_pos % block_size; + write_buf = read_buf; + } + else if (commit_pos / block_size != write_pos / block_size) { + write_buf_start_addr = commit_pos % block_size; + fill_buf (write_buf, write_buf_start_addr); + } + write_pos = commit_pos; +} + +bool zmq::swap_t::empty () +{ + return read_pos == write_pos; +} + +bool zmq::swap_t::full () +{ + return buffer_space () == 1; +} + +void zmq::swap_t::copy_from_file (void *buffer_, size_t count_) +{ + char *dest_ptr = (char *) buffer_; + size_t chunk_size, remainder = count_; + + while (remainder > 0) { + chunk_size = std::min (remainder, + std::min ((size_t) (filesize - read_pos), + (size_t) (block_size - read_pos % block_size))); + + memcpy (dest_ptr, &read_buf [read_pos % block_size], chunk_size); + dest_ptr += chunk_size; + + read_pos = (read_pos + chunk_size) % filesize; + if (read_pos % block_size == 0) { + if (read_pos / block_size == write_pos / block_size) + read_buf = write_buf; + else + fill_buf (read_buf, read_pos); + } + remainder -= chunk_size; + } +} + +void zmq::swap_t::copy_to_file (const void *buffer_, size_t count_) +{ + char *source_ptr = (char *) buffer_; + size_t chunk_size, remainder = count_; + + while (remainder > 0) { + chunk_size = std::min (remainder, + std::min ((size_t) (filesize - write_pos), + (size_t) (block_size - write_pos % block_size))); + + memcpy (&write_buf [write_pos % block_size], source_ptr, chunk_size); + source_ptr += chunk_size; + + write_pos = (write_pos + chunk_size) % filesize; + if (write_pos % block_size == 0) { + save_write_buf (); + write_buf_start_addr = write_pos; + + if (write_buf == read_buf) { + if (read_buf == buf2) + write_buf = buf1; + else + write_buf = buf2; + } + } + remainder -= chunk_size; + } +} + +void zmq::swap_t::fill_buf (char *buf, int64_t pos) +{ + if (file_pos != pos) { +#ifdef ZMQ_HAVE_WINDOWS + __int64 offset = _lseeki64 (fd, pos, SEEK_SET); +#else + off_t offset = lseek (fd, (off_t) pos, SEEK_SET); +#endif + errno_assert (offset == pos); + file_pos = pos; + } + size_t octets_stored = 0; + size_t octets_total = std::min (block_size, (size_t) (filesize - file_pos)); + + while (octets_stored < octets_total) { +#ifdef ZMQ_HAVE_WINDOWS + int rc = _read (fd, &buf [octets_stored], octets_total - octets_stored); +#else + ssize_t rc = read (fd, &buf [octets_stored], octets_total - octets_stored); +#endif + errno_assert (rc > 0); + octets_stored += rc; + } + file_pos += octets_total; +} + +void zmq::swap_t::save_write_buf () +{ + if (file_pos != write_buf_start_addr) { +#ifdef ZMQ_HAVE_WINDOWS + __int64 offset = _lseeki64 (fd, write_buf_start_addr, SEEK_SET); +#else + off_t offset = lseek (fd, (off_t) write_buf_start_addr, SEEK_SET); +#endif + errno_assert (offset == write_buf_start_addr); + file_pos = write_buf_start_addr; + } + size_t octets_stored = 0; + size_t octets_total = std::min (block_size, (size_t) (filesize - file_pos)); + + while (octets_stored < octets_total) { +#ifdef ZMQ_HAVE_WINDOWS + int rc = _write (fd, &write_buf [octets_stored], octets_total - octets_stored); +#else + ssize_t rc = write (fd, &write_buf [octets_stored], octets_total - octets_stored); +#endif + errno_assert (rc > 0); + octets_stored += rc; + } + file_pos += octets_total; +} + +int64_t zmq::swap_t::buffer_space () +{ + if (write_pos < read_pos) + return read_pos - write_pos; + + return filesize - (write_pos - read_pos); +} diff --git a/src/swap.hpp b/src/swap.hpp new file mode 100644 index 0000000..c9dfd99 --- /dev/null +++ b/src/swap.hpp @@ -0,0 +1,114 @@ +/* + Copyright (c) 2007-2010 iMatix Corporation + + This file is part of 0MQ. + + 0MQ is free software; you can redistribute it and/or modify it under + the terms of the Lesser GNU General Public License as published by + the Free Software Foundation; either version 3 of the License, or + (at your option) any later version. + + 0MQ is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + Lesser GNU General Public License for more details. + + You should have received a copy of the Lesser GNU General Public License + along with this program. If not, see . +*/ + +#ifndef __ZMQ_SWAP_HPP_INCLUDED__ +#define __ZMQ_SWAP_HPP_INCLUDED__ + +#include "../include/zmq.h" + +#include +#include "stdint.hpp" + +namespace zmq +{ + + // This class implements a message swap. Messages are retrieved from + // the swap in the same order as they entered it. + + class swap_t + { + public: + + enum { default_block_size = 8192 }; + + // Creates the swap. + swap_t (int64_t filesize_, size_t block_size_ = default_block_size); + + ~swap_t (); + + int init (); + + // Stores the message into the swap. The function + // returns false if the swap is full; true otherwise. + bool store (zmq_msg_t *msg_); + + // Fetches the oldest message from the swap. It is an error + // to call this function when the swap is empty. + void fetch (zmq_msg_t *msg_); + + void commit (); + + void rollback (); + + // Returns true if the swap is empty; false otherwise. + bool empty (); + + // Returns true if and only if the swap is full. + bool full (); + + private: + + // Copies data from a memory buffer to the backing file. + // Wraps around when reaching maximum file size. + void copy_from_file (void *buffer_, size_t count_); + + // Copies data from the backing file to the memory buffer. + // Wraps around when reaching end-of-file. + void copy_to_file (const void *buffer_, size_t count_); + + // Returns the buffer space available. + int64_t buffer_space (); + + void fill_buf (char *buf, int64_t pos); + + void save_write_buf (); + + // File descriptor to the backing file. + int fd; + + // Name of the backing file. + std::string filename; + + // Maximum size of the backing file. + int64_t filesize; + + // File offset associated with the fd file descriptor. + int64_t file_pos; + + // File offset the next message will be stored at. + int64_t write_pos; + + // File offset the next message will be read from. + int64_t read_pos; + + int64_t commit_pos; + + size_t block_size; + + char *buf1; + char *buf2; + char *read_buf; + char *write_buf; + + int64_t write_buf_start_addr; + }; + +} + +#endif -- cgit v1.2.3