Skip to content
Newer
Older
100644 324 lines (268 sloc) 8.33 KB
fca2e8e @hurtonm Add SWAP support
hurtonm authored
1 /*
b7a20bd @sustrik The copyrights in file headers updated.
sustrik authored
2 Copyright (c) 2007-2011 iMatix Corporation
3 Copyright (c) 2007-2011 Other contributors as noted in the AUTHORS file
fca2e8e @hurtonm Add SWAP support
hurtonm authored
4
5 This file is part of 0MQ.
6
7 0MQ is free software; you can redistribute it and/or modify it under
b358df9 @sustrik Name of "GNU Lesser Public License" corrected.
sustrik authored
8 the terms of the GNU Lesser General Public License as published by
fca2e8e @hurtonm Add SWAP support
hurtonm authored
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
b358df9 @sustrik Name of "GNU Lesser Public License" corrected.
sustrik authored
15 GNU Lesser General Public License for more details.
fca2e8e @hurtonm Add SWAP support
hurtonm authored
16
b358df9 @sustrik Name of "GNU Lesser Public License" corrected.
sustrik authored
17 You should have received a copy of the GNU Lesser General Public License
fca2e8e @hurtonm Add SWAP support
hurtonm authored
18 along with this program. If not, see <http://www.gnu.org/licenses/>.
19 */
20
79a3d07 @sustrik MSVC build fixed
sustrik authored
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
fca2e8e @hurtonm Add SWAP support
hurtonm authored
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
d8b975f @sustrik msg_store_t renamed to swap_t
sustrik authored
39 #include "swap.hpp"
beea535 @sustrik swap's block size made into a compile-time parameter
sustrik authored
40 #include "config.hpp"
fca2e8e @hurtonm Add SWAP support
hurtonm authored
41 #include "atomic_counter.hpp"
42 #include "err.hpp"
43
beea535 @sustrik swap's block size made into a compile-time parameter
sustrik authored
44 zmq::swap_t::swap_t (int64_t filesize_) :
fca2e8e @hurtonm Add SWAP support
hurtonm authored
45 fd (-1),
46 filesize (filesize_),
47 file_pos (0),
48 write_pos (0),
49 read_pos (0),
beea535 @sustrik swap's block size made into a compile-time parameter
sustrik authored
50 block_size (swap_block_size),
fca2e8e @hurtonm Add SWAP support
hurtonm authored
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];
43e8868 @sustrik Added explicit error message in case of memory exhaustion
sustrik authored
57 alloc_assert (buf1);
fca2e8e @hurtonm Add SWAP support
hurtonm authored
58
59 buf2 = new (std::nothrow) char [block_size];
43e8868 @sustrik Added explicit error message in case of memory exhaustion
sustrik authored
60 alloc_assert (buf2);
fca2e8e @hurtonm Add SWAP support
hurtonm authored
61
62 read_buf = write_buf = buf1;
63 }
64
d8b975f @sustrik msg_store_t renamed to swap_t
sustrik authored
65 zmq::swap_t::~swap_t ()
fca2e8e @hurtonm Add SWAP support
hurtonm authored
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
d8b975f @sustrik msg_store_t renamed to swap_t
sustrik authored
88 int zmq::swap_t::init ()
fca2e8e @hurtonm Add SWAP support
hurtonm authored
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
7b22dcc Added Android support
Mikko Koppanen authored
114 #if (defined (ZMQ_HAVE_LINUX) && !defined (ZMQ_HAVE_ANDROID))
fca2e8e @hurtonm Add SWAP support
hurtonm authored
115 // Enable more aggresive read-ahead optimization.
116 posix_fadvise (fd, 0, filesize, POSIX_FADV_SEQUENTIAL);
117 #endif
118 return 0;
119 }
120
d8b975f @sustrik msg_store_t renamed to swap_t
sustrik authored
121 bool zmq::swap_t::store (zmq_msg_t *msg_)
fca2e8e @hurtonm Add SWAP support
hurtonm authored
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
d8b975f @sustrik msg_store_t renamed to swap_t
sustrik authored
141 void zmq::swap_t::fetch (zmq_msg_t *msg_)
fca2e8e @hurtonm Add SWAP support
hurtonm authored
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
d8b975f @sustrik msg_store_t renamed to swap_t
sustrik authored
160 void zmq::swap_t::commit ()
fca2e8e @hurtonm Add SWAP support
hurtonm authored
161 {
162 commit_pos = write_pos;
163 }
164
d8b975f @sustrik msg_store_t renamed to swap_t
sustrik authored
165 void zmq::swap_t::rollback ()
fca2e8e @hurtonm Add SWAP support
hurtonm authored
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
d8b975f @sustrik msg_store_t renamed to swap_t
sustrik authored
186 bool zmq::swap_t::empty ()
fca2e8e @hurtonm Add SWAP support
hurtonm authored
187 {
188 return read_pos == write_pos;
189 }
190
27e83cc Fixes assertion on pipe.cpp:237 when swap fills up.
Mikko Koppanen authored
191 /*
d8b975f @sustrik msg_store_t renamed to swap_t
sustrik authored
192 bool zmq::swap_t::full ()
fca2e8e @hurtonm Add SWAP support
hurtonm authored
193 {
27e83cc Fixes assertion on pipe.cpp:237 when swap fills up.
Mikko Koppanen authored
194 // Check that at least the message size can be written to the swap.
195 return buffer_space () < (int64_t) (sizeof (size_t) + 1);
fca2e8e @hurtonm Add SWAP support
hurtonm authored
196 }
27e83cc Fixes assertion on pipe.cpp:237 when swap fills up.
Mikko Koppanen authored
197 */
198
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 }
fca2e8e @hurtonm Add SWAP support
hurtonm authored
208
d8b975f @sustrik msg_store_t renamed to swap_t
sustrik authored
209 void zmq::swap_t::copy_from_file (void *buffer_, size_t count_)
fca2e8e @hurtonm Add SWAP support
hurtonm authored
210 {
1dda8a2 @hintjens Used more expressive variable names
hintjens authored
211 char *dest_ptr = (char *) buffer_;
212 size_t chunk_size, remainder = count_;
fca2e8e @hurtonm Add SWAP support
hurtonm authored
213
1dda8a2 @hintjens Used more expressive variable names
hintjens authored
214 while (remainder > 0) {
ad6bf4a @hintjens Provisional Android support, from Ghislain Putois
hintjens authored
215 chunk_size = std::min (remainder,
1dda8a2 @hintjens Used more expressive variable names
hintjens authored
216 std::min ((size_t) (filesize - read_pos),
fca2e8e @hurtonm Add SWAP support
hurtonm authored
217 (size_t) (block_size - read_pos % block_size)));
218
1dda8a2 @hintjens Used more expressive variable names
hintjens authored
219 memcpy (dest_ptr, &read_buf [read_pos % block_size], chunk_size);
220 dest_ptr += chunk_size;
fca2e8e @hurtonm Add SWAP support
hurtonm authored
221
1dda8a2 @hintjens Used more expressive variable names
hintjens authored
222 read_pos = (read_pos + chunk_size) % filesize;
fca2e8e @hurtonm Add SWAP support
hurtonm authored
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 }
1dda8a2 @hintjens Used more expressive variable names
hintjens authored
229 remainder -= chunk_size;
fca2e8e @hurtonm Add SWAP support
hurtonm authored
230 }
231 }
232
d8b975f @sustrik msg_store_t renamed to swap_t
sustrik authored
233 void zmq::swap_t::copy_to_file (const void *buffer_, size_t count_)
fca2e8e @hurtonm Add SWAP support
hurtonm authored
234 {
1dda8a2 @hintjens Used more expressive variable names
hintjens authored
235 char *source_ptr = (char *) buffer_;
236 size_t chunk_size, remainder = count_;
fca2e8e @hurtonm Add SWAP support
hurtonm authored
237
1dda8a2 @hintjens Used more expressive variable names
hintjens authored
238 while (remainder > 0) {
ad6bf4a @hintjens Provisional Android support, from Ghislain Putois
hintjens authored
239 chunk_size = std::min (remainder,
1dda8a2 @hintjens Used more expressive variable names
hintjens authored
240 std::min ((size_t) (filesize - write_pos),
fca2e8e @hurtonm Add SWAP support
hurtonm authored
241 (size_t) (block_size - write_pos % block_size)));
242
1dda8a2 @hintjens Used more expressive variable names
hintjens authored
243 memcpy (&write_buf [write_pos % block_size], source_ptr, chunk_size);
244 source_ptr += chunk_size;
fca2e8e @hurtonm Add SWAP support
hurtonm authored
245
1dda8a2 @hintjens Used more expressive variable names
hintjens authored
246 write_pos = (write_pos + chunk_size) % filesize;
fca2e8e @hurtonm Add SWAP support
hurtonm authored
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 }
1dda8a2 @hintjens Used more expressive variable names
hintjens authored
258 remainder -= chunk_size;
fca2e8e @hurtonm Add SWAP support
hurtonm authored
259 }
260 }
261
d8b975f @sustrik msg_store_t renamed to swap_t
sustrik authored
262 void zmq::swap_t::fill_buf (char *buf, int64_t pos)
fca2e8e @hurtonm Add SWAP support
hurtonm authored
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 }
1dda8a2 @hintjens Used more expressive variable names
hintjens authored
273 size_t octets_stored = 0;
274 size_t octets_total = std::min (block_size, (size_t) (filesize - file_pos));
fca2e8e @hurtonm Add SWAP support
hurtonm authored
275
1dda8a2 @hintjens Used more expressive variable names
hintjens authored
276 while (octets_stored < octets_total) {
fca2e8e @hurtonm Add SWAP support
hurtonm authored
277 #ifdef ZMQ_HAVE_WINDOWS
1dda8a2 @hintjens Used more expressive variable names
hintjens authored
278 int rc = _read (fd, &buf [octets_stored], octets_total - octets_stored);
fca2e8e @hurtonm Add SWAP support
hurtonm authored
279 #else
43e8868 @sustrik Added explicit error message in case of memory exhaustion
sustrik authored
280 ssize_t rc = read (fd, &buf [octets_stored],
281 octets_total - octets_stored);
fca2e8e @hurtonm Add SWAP support
hurtonm authored
282 #endif
283 errno_assert (rc > 0);
1dda8a2 @hintjens Used more expressive variable names
hintjens authored
284 octets_stored += rc;
fca2e8e @hurtonm Add SWAP support
hurtonm authored
285 }
1dda8a2 @hintjens Used more expressive variable names
hintjens authored
286 file_pos += octets_total;
fca2e8e @hurtonm Add SWAP support
hurtonm authored
287 }
288
d8b975f @sustrik msg_store_t renamed to swap_t
sustrik authored
289 void zmq::swap_t::save_write_buf ()
fca2e8e @hurtonm Add SWAP support
hurtonm authored
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 }
1dda8a2 @hintjens Used more expressive variable names
hintjens authored
300 size_t octets_stored = 0;
301 size_t octets_total = std::min (block_size, (size_t) (filesize - file_pos));
fca2e8e @hurtonm Add SWAP support
hurtonm authored
302
1dda8a2 @hintjens Used more expressive variable names
hintjens authored
303 while (octets_stored < octets_total) {
fca2e8e @hurtonm Add SWAP support
hurtonm authored
304 #ifdef ZMQ_HAVE_WINDOWS
43e8868 @sustrik Added explicit error message in case of memory exhaustion
sustrik authored
305 int rc = _write (fd, &write_buf [octets_stored],
306 octets_total - octets_stored);
fca2e8e @hurtonm Add SWAP support
hurtonm authored
307 #else
43e8868 @sustrik Added explicit error message in case of memory exhaustion
sustrik authored
308 ssize_t rc = write (fd, &write_buf [octets_stored],
309 octets_total - octets_stored);
fca2e8e @hurtonm Add SWAP support
hurtonm authored
310 #endif
311 errno_assert (rc > 0);
1dda8a2 @hintjens Used more expressive variable names
hintjens authored
312 octets_stored += rc;
fca2e8e @hurtonm Add SWAP support
hurtonm authored
313 }
1dda8a2 @hintjens Used more expressive variable names
hintjens authored
314 file_pos += octets_total;
fca2e8e @hurtonm Add SWAP support
hurtonm authored
315 }
316
d8b975f @sustrik msg_store_t renamed to swap_t
sustrik authored
317 int64_t zmq::swap_t::buffer_space ()
fca2e8e @hurtonm Add SWAP support
hurtonm authored
318 {
319 if (write_pos < read_pos)
320 return read_pos - write_pos;
321
322 return filesize - (write_pos - read_pos);
323 }
Something went wrong with that request. Please try again.