1 /*
2 Copyright (c) 2007-2011 iMatix Corporation
3 Copyright (c) 2007-2011 Other contributors as noted in the AUTHORS file
4
5 This file is part of 0MQ.
6
7 0MQ is free software; you can redistribute it and/or modify it under
8 the terms of the GNU Lesser General Public License as published by
9 the Free Software Foundation; either version 3 of the License, or
10 (at your option) any later version.
11
12 0MQ is distributed in the hope that it will be useful,
13 but WITHOUT ANY WARRANTY; without even the implied warranty of
14 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15 GNU Lesser General Public License for more details.
16
17 You should have received a copy of the GNU Lesser General Public License
18 along with this program. If not, see <http://www.gnu.org/licenses/>.
19 */
20
21 #include "platform.hpp"
22
23 #ifdef ZMQ_HAVE_WINDOWS
24 #include "windows.hpp"
25 #include <io.h>
26 #else
27 #include <unistd.h>
28 #endif
29
30 #include "../include/zmq.h"
31
32 #include <sys/types.h>
33 #include <sys/stat.h>
34 #include <fcntl.h>
35 #include <string.h>
36 #include <sstream>
37 #include <algorithm>
38
39 #include "swap.hpp"
40 #include "config.hpp"
41 #include "atomic_counter.hpp"
42 #include "err.hpp"
43
swap_t(int64_t filesize_)44 zmq::swap_t::swap_t (int64_t filesize_) :
45 fd (-1),
46 filesize (filesize_),
47 file_pos (0),
48 write_pos (0),
49 read_pos (0),
50 block_size (swap_block_size),
51 write_buf_start_addr (0)
52 {
53 zmq_assert (filesize > 0);
54 zmq_assert (block_size > 0);
55
56 buf1 = new (std::nothrow) char [block_size];
57 alloc_assert (buf1);
58
59 buf2 = new (std::nothrow) char [block_size];
60 alloc_assert (buf2);
61
62 read_buf = write_buf = buf1;
63 }
64
~swap_t()65 zmq::swap_t::~swap_t ()
66 {
67 delete [] buf1;
68 delete [] buf2;
69
70 if (fd == -1)
71 return;
72
73 #ifdef ZMQ_HAVE_WINDOWS
74 int rc = _close (fd);
75 #else
76 int rc = close (fd);
77 #endif
78 errno_assert (rc == 0);
79
80 #ifdef ZMQ_HAVE_WINDOWS
81 rc = _unlink (filename.c_str ());
82 #else
83 rc = unlink (filename.c_str ());
84 #endif
85 errno_assert (rc == 0);
86 }
87
init()88 int zmq::swap_t::init ()
89 {
90 static zmq::atomic_counter_t seqnum (0);
91
92 // Get process ID.
93 #ifdef ZMQ_HAVE_WINDOWS
94 int pid = GetCurrentThreadId ();
95 #else
96 pid_t pid = getpid ();
97 #endif
98
99 std::ostringstream outs;
100 outs << "zmq_" << pid << '_' << seqnum.get () << ".swap";
101 filename = outs.str ();
102
103 seqnum.add (1);
104
105 // Open the backing file.
106 #ifdef ZMQ_HAVE_WINDOWS
107 fd = _open (filename.c_str (), _O_RDWR | _O_CREAT, 0600);
108 #else
109 fd = open (filename.c_str (), O_RDWR | O_CREAT, 0600);
110 #endif
111 if (fd == -1)
112 return -1;
113
114 #if (defined (ZMQ_HAVE_LINUX) && !defined (ZMQ_HAVE_ANDROID))
115 // Enable more aggresive read-ahead optimization.
116 posix_fadvise (fd, 0, filesize, POSIX_FADV_SEQUENTIAL);
117 #endif
118 return 0;
119 }
120
store(zmq_msg_t * msg_)121 bool zmq::swap_t::store (zmq_msg_t *msg_)
122 {
123 size_t msg_size = zmq_msg_size (msg_);
124
125 // Check buffer space availability.
126 // NOTE: We always keep one byte open.
127 if (buffer_space () <= (int64_t) (sizeof msg_size + 1 + msg_size))
128 return false;
129
130 // Don't store the ZMQ_MSG_SHARED flag.
131 uint8_t msg_flags = msg_->flags & ~ZMQ_MSG_SHARED;
132
133 // Write message length, flags, and message body.
134 copy_to_file (&msg_size, sizeof msg_size);
135 copy_to_file (&msg_flags, sizeof msg_flags);
136 copy_to_file (zmq_msg_data (msg_), msg_size);
137
138 return true;
139 }
140
fetch(zmq_msg_t * msg_)141 void zmq::swap_t::fetch (zmq_msg_t *msg_)
142 {
143 // There must be at least one message available.
144 zmq_assert (read_pos != write_pos);
145
146 // Retrieve the message size.
147 size_t msg_size;
148 copy_from_file (&msg_size, sizeof msg_size);
149
150 // Initialize the message.
151 zmq_msg_init_size (msg_, msg_size);
152
153 // Retrieve the message flags.
154 copy_from_file (&msg_->flags, sizeof msg_->flags);
155
156 // Retrieve the message payload.
157 copy_from_file (zmq_msg_data (msg_), msg_size);
158 }
159
commit()160 void zmq::swap_t::commit ()
161 {
162 commit_pos = write_pos;
163 }
164
rollback()165 void zmq::swap_t::rollback ()
166 {
167 if (commit_pos == write_pos || read_pos == write_pos)
168 return;
169
170 if (write_pos > read_pos)
171 zmq_assert (read_pos <= commit_pos && commit_pos <= write_pos);
172 else
173 zmq_assert (read_pos <= commit_pos || commit_pos <= write_pos);
174
175 if (commit_pos / block_size == read_pos / block_size) {
176 write_buf_start_addr = commit_pos % block_size;
177 write_buf = read_buf;
178 }
179 else if (commit_pos / block_size != write_pos / block_size) {
180 write_buf_start_addr = commit_pos % block_size;
181 fill_buf (write_buf, write_buf_start_addr);
182 }
183 write_pos = commit_pos;
184 }
185
empty()186 bool zmq::swap_t::empty ()
187 {
188 return read_pos == write_pos;
189 }
190
191 /*
192 bool zmq::swap_t::full ()
193 {
194 // Check that at least the message size can be written to the swap.
195 return buffer_space () < (int64_t) (sizeof (size_t) + 1);
196 }
197 */
198
fits(zmq_msg_t * msg_)199 bool zmq::swap_t::fits (zmq_msg_t *msg_)
200 {
201 // Check whether whole binary representation of the message
202 // fits into the swap.
203 size_t msg_size = zmq_msg_size (msg_);
204 if (buffer_space () <= (int64_t) (sizeof msg_size + 1 + msg_size))
205 return false;
206 return true;
207 }
208
copy_from_file(void * buffer_,size_t count_)209 void zmq::swap_t::copy_from_file (void *buffer_, size_t count_)
210 {
211 char *dest_ptr = (char *) buffer_;
212 size_t chunk_size, remainder = count_;
213
214 while (remainder > 0) {
215 chunk_size = std::min (remainder,
216 std::min ((size_t) (filesize - read_pos),
217 (size_t) (block_size - read_pos % block_size)));
218
219 memcpy (dest_ptr, &read_buf [read_pos % block_size], chunk_size);
220 dest_ptr += chunk_size;
221
222 read_pos = (read_pos + chunk_size) % filesize;
223 if (read_pos % block_size == 0) {
224 if (read_pos / block_size == write_pos / block_size)
225 read_buf = write_buf;
226 else
227 fill_buf (read_buf, read_pos);
228 }
229 remainder -= chunk_size;
230 }
231 }
232
copy_to_file(const void * buffer_,size_t count_)233 void zmq::swap_t::copy_to_file (const void *buffer_, size_t count_)
234 {
235 char *source_ptr = (char *) buffer_;
236 size_t chunk_size, remainder = count_;
237
238 while (remainder > 0) {
239 chunk_size = std::min (remainder,
240 std::min ((size_t) (filesize - write_pos),
241 (size_t) (block_size - write_pos % block_size)));
242
243 memcpy (&write_buf [write_pos % block_size], source_ptr, chunk_size);
244 source_ptr += chunk_size;
245
246 write_pos = (write_pos + chunk_size) % filesize;
247 if (write_pos % block_size == 0) {
248 save_write_buf ();
249 write_buf_start_addr = write_pos;
250
251 if (write_buf == read_buf) {
252 if (read_buf == buf2)
253 write_buf = buf1;
254 else
255 write_buf = buf2;
256 }
257 }
258 remainder -= chunk_size;
259 }
260 }
261
fill_buf(char * buf,int64_t pos)262 void zmq::swap_t::fill_buf (char *buf, int64_t pos)
263 {
264 if (file_pos != pos) {
265 #ifdef ZMQ_HAVE_WINDOWS
266 __int64 offset = _lseeki64 (fd, pos, SEEK_SET);
267 #else
268 off_t offset = lseek (fd, (off_t) pos, SEEK_SET);
269 #endif
270 errno_assert (offset == pos);
271 file_pos = pos;
272 }
273 size_t octets_stored = 0;
274 size_t octets_total = std::min (block_size, (size_t) (filesize - file_pos));
275
276 while (octets_stored < octets_total) {
277 #ifdef ZMQ_HAVE_WINDOWS
278 int rc = _read (fd, &buf [octets_stored], octets_total - octets_stored);
279 #else
280 ssize_t rc = read (fd, &buf [octets_stored],
281 octets_total - octets_stored);
282 #endif
283 errno_assert (rc > 0);
284 octets_stored += rc;
285 }
286 file_pos += octets_total;
287 }
288
save_write_buf()289 void zmq::swap_t::save_write_buf ()
290 {
291 if (file_pos != write_buf_start_addr) {
292 #ifdef ZMQ_HAVE_WINDOWS
293 __int64 offset = _lseeki64 (fd, write_buf_start_addr, SEEK_SET);
294 #else
295 off_t offset = lseek (fd, (off_t) write_buf_start_addr, SEEK_SET);
296 #endif
297 errno_assert (offset == write_buf_start_addr);
298 file_pos = write_buf_start_addr;
299 }
300 size_t octets_stored = 0;
301 size_t octets_total = std::min (block_size, (size_t) (filesize - file_pos));
302
303 while (octets_stored < octets_total) {
304 #ifdef ZMQ_HAVE_WINDOWS
305 int rc = _write (fd, &write_buf [octets_stored],
306 octets_total - octets_stored);
307 #else
308 ssize_t rc = write (fd, &write_buf [octets_stored],
309 octets_total - octets_stored);
310 #endif
311 errno_assert (rc > 0);
312 octets_stored += rc;
313 }
314 file_pos += octets_total;
315 }
316
buffer_space()317 int64_t zmq::swap_t::buffer_space ()
318 {
319 if (write_pos < read_pos)
320 return read_pos - write_pos;
321
322 return filesize - (write_pos - read_pos);
323 }
324