1 /*++
2 /* NAME
3 /* vstream 3
4 /* SUMMARY
5 /* light-weight buffered I/O package
6 /* SYNOPSIS
7 /* #include <vstream.h>
8 /*
9 /* VSTREAM *vstream_fopen(path, flags, mode)
10 /* const char *path;
11 /* int flags;
12 /* mode_t mode;
13 /*
14 /* VSTREAM *vstream_fdopen(fd, flags)
15 /* int fd;
16 /* int flags;
17 /*
18 /* VSTREAM *vstream_memopen(string, flags)
19 /* VSTRING *string;
20 /* int flags;
21 /*
22 /* VSTREAM *vstream_memreopen(stream, string, flags)
23 /* VSTREAM *stream;
24 /* VSTRING *string;
25 /* int flags;
26 /*
27 /* int vstream_fclose(stream)
28 /* VSTREAM *stream;
29 /*
30 /* int vstream_fdclose(stream)
31 /* VSTREAM *stream;
32 /*
33 /* VSTREAM *vstream_printf(format, ...)
34 /* const char *format;
35 /*
36 /* VSTREAM *vstream_fprintf(stream, format, ...)
37 /* VSTREAM *stream;
38 /* const char *format;
39 /*
40 /* int VSTREAM_GETC(stream)
41 /* VSTREAM *stream;
42 /*
43 /* int VSTREAM_PUTC(ch, stream)
44 /* int ch;
45 /*
46 /* int VSTREAM_GETCHAR(void)
47 /*
48 /* int VSTREAM_PUTCHAR(ch)
49 /* int ch;
50 /*
51 /* int vstream_ungetc(stream, ch)
52 /* VSTREAM *stream;
53 /* int ch;
54 /*
55 /* int vstream_fputs(str, stream)
56 /* const char *str;
57 /* VSTREAM *stream;
58 /*
59 /* off_t vstream_ftell(stream)
60 /* VSTREAM *stream;
61 /*
62 /* off_t vstream_fseek(stream, offset, whence)
63 /* VSTREAM *stream;
64 /* off_t offset;
65 /* int whence;
66 /*
67 /* int vstream_fflush(stream)
68 /* VSTREAM *stream;
69 /*
70 /* int vstream_fpurge(stream, direction)
71 /* VSTREAM *stream;
72 /* int direction;
73 /*
74 /* ssize_t vstream_fread(stream, buf, len)
75 /* VSTREAM *stream;
76 /* void *buf;
77 /* ssize_t len;
78 /*
79 /* ssize_t vstream_fwrite(stream, buf, len)
80 /* VSTREAM *stream;
81 /* void *buf;
82 /* ssize_t len;
83 /*
84 /* ssize_t vstream_fread_app(stream, buf, len)
85 /* VSTREAM *stream;
86 /* VSTRING *buf;
87 /* ssize_t len;
88 /*
89 /* ssize_t vstream_fread_buf(stream, buf, len)
90 /* VSTREAM *stream;
91 /* VSTRING *buf;
92 /* ssize_t len;
93 /*
94 /* void vstream_control(stream, name, ...)
95 /* VSTREAM *stream;
96 /* int name;
97 /*
98 /* int vstream_fileno(stream)
99 /* VSTREAM *stream;
100 /*
101 /* const ssize_t vstream_req_bufsize(stream)
102 /* VSTREAM *stream;
103 /*
104 /* void *vstream_context(stream)
105 /* VSTREAM *stream;
106 /*
107 /* int vstream_ferror(stream)
108 /* VSTREAM *stream;
109 /*
110 /* int vstream_ftimeout(stream)
111 /* VSTREAM *stream;
112 /*
113 /* int vstream_feof(stream)
114 /* VSTREAM *stream;
115 /*
116 /* int vstream_clearerr(stream)
117 /* VSTREAM *stream;
118 /*
119 /* const char *VSTREAM_PATH(stream)
120 /* VSTREAM *stream;
121 /*
122 /* char *vstream_vprintf(format, ap)
123 /* const char *format;
124 /* va_list *ap;
125 /*
126 /* char *vstream_vfprintf(stream, format, ap)
127 /* VSTREAM *stream;
128 /* const char *format;
129 /* va_list *ap;
130 /*
131 /* ssize_t vstream_bufstat(stream, command)
132 /* VSTREAM *stream;
133 /* int command;
134 /*
135 /* ssize_t vstream_peek(stream)
136 /* VSTREAM *stream;
137 /*
138 /* const char *vstream_peek_data(stream)
139 /* VSTREAM *stream;
140 /*
141 /* int vstream_setjmp(stream)
142 /* VSTREAM *stream;
143 /*
144 /* void vstream_longjmp(stream, val)
145 /* VSTREAM *stream;
146 /* int val;
147 /*
148 /* time_t vstream_ftime(stream)
149 /* VSTREAM *stream;
150 /*
151 /* struct timeval vstream_ftimeval(stream)
152 /* VSTREAM *stream;
153 /*
154 /* int vstream_rd_error(stream)
155 /* VSTREAM *stream;
156 /*
157 /* int vstream_wr_error(stream)
158 /* VSTREAM *stream;
159 /*
160 /* int vstream_rd_timeout(stream)
161 /* VSTREAM *stream;
162 /*
163 /* int vstream_wr_timeout(stream)
164 /* VSTREAM *stream;
165 /*
166 /* int vstream_fstat(stream, flags)
167 /* VSTREAM *stream;
168 /* int flags;
169 /* DESCRIPTION
170 /* The \fIvstream\fR module implements light-weight buffered I/O
171 /* similar to the standard I/O routines.
172 /*
173 /* The interface is implemented in terms of VSTREAM structure
174 /* pointers, also called streams. For convenience, three streams
175 /* are predefined: VSTREAM_IN, VSTREAM_OUT, and VSTREAM_ERR. These
176 /* streams are connected to the standard input, output and error
177 /* file descriptors, respectively.
178 /*
179 /* Although the interface is patterned after the standard I/O
180 /* library, there are some major differences:
181 /* .IP \(bu
182 /* File descriptors are not limited to the range 0..255. This
183 /* was reason #1 to write these routines in the first place.
184 /* .IP \(bu
185 /* The application can switch between reading and writing on
186 /* the same stream without having to perform a flush or seek
187 /* operation, and can change write position without having to
188 /* flush. This was reason #2. Upon position or direction change,
189 /* unread input is discarded, and unwritten output is flushed
190 /* automatically. Exception: with double-buffered streams, unread
191 /* input is not discarded upon change of I/O direction, and
192 /* output flushing is delayed until the read buffer must be refilled.
193 /* .IP \(bu
194 /* A bidirectional stream can read and write with the same buffer
195 /* and file descriptor, or it can have separate read/write
196 /* buffers and/or file descriptors.
197 /* .IP \(bu
198 /* No automatic flushing of VSTREAM_OUT upon program exit, or of
199 /* VSTREAM_ERR at any time. No unbuffered or line buffered modes.
200 /* This functionality may be added when it is really needed.
201 /* .PP
202 /* vstream_fopen() opens the named file and associates a buffered
203 /* stream with it. The \fIpath\fR, \fIflags\fR and \fImode\fR
204 /* arguments are passed on to the open(2) routine. The result is
205 /* a null pointer in case of problems. The \fIpath\fR argument is
206 /* copied and can be looked up with VSTREAM_PATH().
207 /*
208 /* vstream_fdopen() takes an open file and associates a buffered
209 /* stream with it. The \fIflags\fR argument specifies how the file
210 /* was opened. vstream_fdopen() either succeeds or never returns.
211 /*
212 /* vstream_memopen() opens a VSTRING as a stream. The \fIflags\fR
213 /* argument must specify one of O_RDONLY, O_WRONLY, or O_APPEND.
214 /* vstream_memopen() either succeeds or never returns. Streams
215 /* opened with vstream_memopen() have limitations: they can't
216 /* be opened in read/write mode, they can't seek beyond the
217 /* end of the VSTRING, and they don't support vstream_control()
218 /* methods that manipulate buffers, file descriptors, or I/O
219 /* functions. After a VSTRING is opened for writing, its content
220 /* will be in an indeterminate state while the stream is open,
221 /* and will be null-terminated when the stream is closed.
222 /*
223 /* vstream_memreopen() reopens a memory stream. When the
224 /* \fIstream\fR argument is a null pointer, the behavior is that
225 /* of vstream_memopen().
226 /*
227 /* vstream_fclose() closes the named buffered stream. The result
228 /* is 0 in case of success, VSTREAM_EOF in case of problems.
229 /* vstream_fclose() reports the same errors as vstream_ferror().
230 /*
231 /* vstream_fdclose() leaves the file(s) open but is otherwise
232 /* identical to vstream_fclose().
233 /*
234 /* vstream_fprintf() formats its arguments according to the
235 /* \fIformat\fR argument and writes the result to the named stream.
236 /* The result is the stream argument. It understands the s, c, d, u,
237 /* o, x, X, e, f and g format types, the l modifier, field width and
238 /* precision, sign, and padding with zeros or spaces. In addition,
239 /* vstream_fprintf() recognizes the %m format specifier and expands
240 /* it to the error message corresponding to the current value of the
241 /* global \fIerrno\fR variable.
242 /*
243 /* vstream_printf() performs formatted output to the standard output
244 /* stream.
245 /*
246 /* VSTREAM_GETC() reads the next character from the named stream.
247 /* The result is VSTREAM_EOF when end-of-file is reached or if a read
248 /* error was detected. VSTREAM_GETC() is an unsafe macro that
249 /* evaluates some arguments more than once.
250 /*
251 /* VSTREAM_GETCHAR() is an alias for VSTREAM_GETC(VSTREAM_IN).
252 /*
253 /* VSTREAM_PUTC() appends the specified character to the specified
254 /* stream. The result is the stored character, or VSTREAM_EOF in
255 /* case of problems. VSTREAM_PUTC() is an unsafe macro that
256 /* evaluates some arguments more than once.
257 /*
258 /* VSTREAM_PUTCHAR(c) is an alias for VSTREAM_PUTC(c, VSTREAM_OUT).
259 /*
260 /* vstream_ungetc() pushes back a character onto the specified stream
261 /* and returns the character, or VSTREAM_EOF in case of problems.
262 /* It is an error to push back before reading (or immediately after
263 /* changing the stream offset via vstream_fseek()). Upon successful
264 /* return, vstream_ungetc() clears the end-of-file stream flag.
265 /*
266 /* vstream_fputs() appends the given null-terminated string to the
267 /* specified buffered stream. The result is 0 in case of success,
268 /* VSTREAM_EOF in case of problems.
269 /*
270 /* vstream_ftell() returns the file offset for the specified stream,
271 /* -1 if the stream is connected to a non-seekable file.
272 /*
273 /* vstream_fseek() changes the file position for the next read or write
274 /* operation. Unwritten output is flushed. With unidirectional streams,
275 /* unread input is discarded. The \fIoffset\fR argument specifies the file
276 /* position from the beginning of the file (\fIwhence\fR is SEEK_SET),
277 /* from the current file position (\fIwhence\fR is SEEK_CUR), or from
278 /* the file end (SEEK_END). The result value is the file offset
279 /* from the beginning of the file, -1 in case of problems.
280 /*
281 /* vstream_fflush() flushes unwritten data to a file that was
282 /* opened in read-write or write-only mode.
283 /* vstream_fflush() returns 0 in case of success, VSTREAM_EOF in
284 /* case of problems. It is an error to flush a read-only stream.
285 /* vstream_fflush() reports the same errors as vstream_ferror().
286 /*
287 /* vstream_fpurge() discards the contents of the stream buffer.
288 /* If direction is VSTREAM_PURGE_READ, it discards unread data,
289 /* else if direction is VSTREAM_PURGE_WRITE, it discards unwritten
290 /* data. In the case of a double-buffered stream, if direction is
291 /* VSTREAM_PURGE_BOTH, it discards the content of both the read
292 /* and write buffers. vstream_fpurge() returns 0 in case of success,
293 /* VSTREAM_EOF in case of problems.
294 /*
295 /* vstream_fread() and vstream_fwrite() perform unformatted I/O
296 /* on the named stream. The result value is the number of bytes
297 /* transferred. A short count is returned in case of end-of-file
298 /* or error conditions.
299 /*
300 /* vstream_fread_buf() resets the buffer write position,
301 /* allocates space for the specified number of bytes in the
302 /* buffer, reads the bytes from the specified VSTREAM, and
303 /* adjusts the buffer write position. The buffer is NOT
304 /* null-terminated. The result value is as with vstream_fread().
305 /* NOTE: do not skip calling vstream_fread_buf() when len == 0.
306 /* This function has side effects including resetting the buffer
307 /* write position, and skipping the call would invalidate the
308 /* buffer state.
309 /*
310 /* vstream_fread_app() is like vstream_fread_buf() but appends
311 /* to existing buffer content, instead of writing over it.
312 /*
313 /* vstream_control() allows the user to fine tune the behavior of
314 /* the specified stream. The arguments are a list of macros with
315 /* zero or more arguments, terminated with CA_VSTREAM_CTL_END
316 /* which has none. The following lists the names and the types
317 /* of the corresponding value arguments.
318 /* .IP "CA_VSTREAM_CTL_READ_FN(ssize_t (*)(int, void *, size_t, int, void *))"
319 /* The argument specifies an alternative for the timed_read(3) function,
320 /* for example, a read function that performs decryption.
321 /* This function receives as arguments a file descriptor, buffer pointer,
322 /* buffer length, timeout value, and the VSTREAM's context value.
323 /* A timeout value <= 0 disables the time limit.
324 /* This function should return the positive number of bytes transferred,
325 /* 0 upon EOF, and -1 upon error with errno set appropriately.
326 /* .IP "CA_VSTREAM_CTL_WRITE_FN(ssize_t (*)(int, void *, size_t, int, void *))"
327 /* The argument specifies an alternative for the timed_write(3) function,
328 /* for example, a write function that performs encryption.
329 /* This function receives as arguments a file descriptor, buffer pointer,
330 /* buffer length, timeout value, and the VSTREAM's context value.
331 /* A timeout value <= 0 disables the time limit.
332 /* This function should return the positive number of bytes transferred,
333 /* and -1 upon error with errno set appropriately. Instead of -1 it may
334 /* also return 0, e.g., upon remote party-initiated protocol shutdown.
335 /* .IP "CA_VSTREAM_CTL_CONTEXT(void *)"
336 /* The argument specifies application context that is passed on to
337 /* the application-specified read/write routines. No copy is made.
338 /* .IP "CA_VSTREAM_CTL_PATH(const char *)"
339 /* Updates the stored pathname of the specified stream. The pathname
340 /* is copied.
341 /* .IP "CA_VSTREAM_CTL_DOUBLE (no arguments)"
342 /* Use separate buffers for reading and for writing. This prevents
343 /* unread input from being discarded upon change of I/O direction.
344 /* .IP "CA_VSTREAM_CTL_READ_FD(int)"
345 /* The argument specifies the file descriptor to be used for reading.
346 /* This feature is limited to double-buffered streams, and makes the
347 /* stream non-seekable.
348 /* .IP "CA_VSTREAM_CTL_WRITE_FD(int)"
349 /* The argument specifies the file descriptor to be used for writing.
350 /* This feature is limited to double-buffered streams, and makes the
351 /* stream non-seekable.
352 /* .IP "CA_VSTREAM_CTL_SWAP_FD(VSTREAM *)"
353 /* The argument specifies a VSTREAM pointer; the request swaps the
354 /* file descriptor members of the two streams. This feature is limited
355 /* to streams that are both double-buffered or both single-buffered.
356 /* .IP "CA_VSTREAM_CTL_DUPFD(int)"
357 /* The argument specifies a minimum file descriptor value. If
358 /* the actual stream's file descriptors are below the minimum,
359 /* reallocate the descriptors to the first free value greater
360 /* than or equal to the minimum. The VSTREAM_CTL_DUPFD macro
361 /* is defined only on systems with fcntl() F_DUPFD support.
362 /* .IP "CA_VSTREAM_CTL_WAITPID_FN(int (*)(pid_t, WAIT_STATUS_T *, int))"
363 /* A pointer to function that behaves like waitpid(). This information
364 /* is used by the vstream_pclose() routine.
365 /* .IP "CA_VSTREAM_CTL_TIMEOUT(int)"
366 /* The deadline for a descriptor to become readable in case of a read
367 /* request, or writable in case of a write request. Specify a value
368 /* of 0 to disable deadlines.
369 /* .IP "CA_VSTREAM_CTL_EXCEPT (no arguments)"
370 /* Enable exception handling with vstream_setjmp() and vstream_longjmp().
371 /* This involves allocation of additional memory that normally isn't
372 /* used.
373 /* .IP "CA_VSTREAM_CTL_BUFSIZE(ssize_t)"
374 /* Specify a non-default buffer size for the next read(2) or
375 /* write(2) operation, or zero to implement a no-op. Requests
376 /* to reduce the buffer size are silently ignored (i.e. any
377 /* positive value <= vstream_req_bufsize()). To get a buffer
378 /* size smaller than VSTREAM_BUFSIZE, make the VSTREAM_CTL_BUFSIZE
379 /* request before the first stream read or write operation
380 /* (i.e., vstream_req_bufsize() returns zero). Requests to
381 /* change a fixed-size buffer (i.e., VSTREAM_ERR) are not
382 /* allowed.
383 /*
384 /* NOTE: the vstream_*printf() routines may silently expand a
385 /* buffer, so that the result of some %letter specifiers can
386 /* be written to contiguous memory.
387 /* .IP CA_VSTREAM_CTL_START_DEADLINE (no arguments)
388 /* Change the VSTREAM_CTL_TIMEOUT behavior, to a deadline for
389 /* the total amount of time for all subsequent file descriptor
390 /* read or write operations, and recharge the deadline timer.
391 /* .IP CA_VSTREAM_CTL_STOP_DEADLINE (no arguments)
392 /* Revert VSTREAM_CTL_TIMEOUT behavior to the default, i.e.
393 /* a time limit for individual file descriptor read or write
394 /* operations.
395 /* .IP CA_VSTREAM_CTL_MIN_DATA_RATE (int)
396 /* When the DEADLINE is enabled, the amount of data that must
397 /* be transferred to add 1 second to the deadline. However,
398 /* the deadline will never exceed the timeout specified with
399 /* VSTREAM_CTL_TIMEOUT. A zero value requests no update to the
400 /* deadline as data is transferred; that is appropriate for
401 /* request/reply interactions.
402 /* .IP CA_VSTREAM_CTL_OWN_VSTRING (no arguments)
403 /* Transfer ownership of the VSTRING that was opened with
404 /* vstream_memopen() etc. to the stream, so that the VSTRING
405 /* is automatically destroyed when the stream is closed.
406 /* .PP
407 /* vstream_fileno() gives access to the file handle associated with
408 /* a buffered stream. With streams that have separate read/write
409 /* file descriptors, the result is the current descriptor.
410 /*
411 /* vstream_req_bufsize() returns the buffer size that will be
412 /* used for the next read(2) or write(2) operation on the named
413 /* stream. A zero result means that the next read(2) or write(2)
414 /* operation will use the default buffer size (VSTREAM_BUFSIZE).
415 /*
416 /* vstream_context() returns the application context that is passed on to
417 /* the application-specified read/write routines.
418 /*
419 /* VSTREAM_PATH() is an unsafe macro that returns the name stored
420 /* with vstream_fopen() or with vstream_control(). The macro is
421 /* unsafe because it evaluates some arguments more than once.
422 /*
423 /* vstream_feof() returns non-zero when a previous operation on the
424 /* specified stream caused an end-of-file condition.
425 /* Although further read requests after EOF may complete
426 /* successfully, vstream_feof() will keep returning non-zero
427 /* until vstream_clearerr() is called for that stream.
428 /*
429 /* vstream_ferror() returns non-zero when a previous operation on the
430 /* specified stream caused a non-EOF error condition, including timeout.
431 /* After a non-EOF error on a stream, no I/O request will
432 /* complete until after vstream_clearerr() is called for that stream.
433 /*
434 /* vstream_ftimeout() returns non-zero when a previous operation on the
435 /* specified stream caused a timeout error condition. See
436 /* vstream_ferror() for error persistence details.
437 /*
438 /* vstream_clearerr() resets the timeout, error and end-of-file indication
439 /* of the specified stream, and returns no useful result.
440 /*
441 /* vstream_vfprintf() provides an alternate interface
442 /* for formatting an argument list according to a format string.
443 /*
444 /* vstream_vprintf() provides a similar alternative interface.
445 /*
446 /* vstream_bufstat() provides input and output buffer status
447 /* information. The command is one of the following:
448 /* .IP VSTREAM_BST_IN_PEND
449 /* Return the number of characters that can be read without
450 /* refilling the read buffer.
451 /* .IP VSTREAM_BST_OUT_PEND
452 /* Return the number of characters that are waiting in the
453 /* write buffer.
454 /* .PP
455 /* vstream_peek() returns the number of characters that can be
456 /* read from the named stream without refilling the read buffer.
457 /* This is an alias for vstream_bufstat(stream, VSTREAM_BST_IN_PEND).
458 /*
459 /* vstream_peek_data() returns a pointer to the unread bytes
460 /* that exist according to vstream_peek(), or null if no unread
461 /* bytes are available.
462 /*
463 /* vstream_setjmp() saves processing context and makes that context
464 /* available for use with vstream_longjmp(). Normally, vstream_setjmp()
465 /* returns zero. A non-zero result means that vstream_setjmp() returned
466 /* through a vstream_longjmp() call; the result is the \fIval\fR argument
467 /* given to vstream_longjmp().
468 /*
469 /* NB: non-local jumps such as vstream_longjmp() are not safe
470 /* for jumping out of any routine that manipulates VSTREAM data.
471 /* longjmp() like calls are best avoided in signal handlers.
472 /*
473 /* vstream_ftime() returns the time of initialization, the last buffer
474 /* fill operation, or the last buffer flush operation for the specified
475 /* stream. This information is maintained only when stream timeouts are
476 /* enabled.
477 /*
478 /* vstream_ftimeval() is like vstream_ftime() but returns more
479 /* detail.
480 /*
481 /* vstream_rd_mumble() and vstream_wr_mumble() report on
482 /* read and write error conditions, respectively.
483 /*
484 /* vstream_fstat() queries stream status information about
485 /* user-requested features. The \fIflags\fR argument is the
486 /* bitwise OR of one or more of the following, and the result
487 /* value is the bitwise OR of the features that are activated.
488 /* .IP VSTREAM_FLAG_DEADLINE
489 /* The deadline feature is activated.
490 /* .IP VSTREAM_FLAG_DOUBLE
491 /* The double-buffering feature is activated.
492 /* .IP VSTREAM_FLAG_MEMORY
493 /* The stream is connected to a VSTRING buffer.
494 /* .IP VSTREAM_FLAG_OWN_VSTRING
495 /* The stream 'owns' the VSTRING buffer, and is responsible
496 /* for cleaning up when the stream is closed.
497 /* DIAGNOSTICS
498 /* Panics: interface violations. Fatal errors: out of memory.
499 /* SEE ALSO
500 /* timed_read(3) default read routine
501 /* timed_write(3) default write routine
502 /* vbuf_print(3) formatting engine
503 /* setjmp(3) non-local jumps
504 /* BUGS
505 /* Should use mmap() on reasonable systems.
506 /* LICENSE
507 /* .ad
508 /* .fi
509 /* The Secure Mailer license must be distributed with this software.
510 /* AUTHOR(S)
511 /* Wietse Venema
512 /* IBM T.J. Watson Research
513 /* P.O. Box 704
514 /* Yorktown Heights, NY 10598, USA
515 /*
516 /* Wietse Venema
517 /* Google, Inc.
518 /* 111 8th Avenue
519 /* New York, NY 10011, USA
520 /*--*/
521
522 /* System library. */
523
524 #include <sys_defs.h>
525 #include <stdlib.h> /* 44BSD stdarg.h uses abort() */
526 #include <stdarg.h>
527 #include <stddef.h>
528 #include <unistd.h>
529 #include <fcntl.h>
530 #include <time.h>
531 #include <errno.h>
532 #include <string.h>
533
534 /* Utility library. */
535
536 #define VSTRING_INTERNAL
537
538 #include "mymalloc.h"
539 #include "msg.h"
540 #include "vbuf_print.h"
541 #include "iostuff.h"
542 #include "vstring.h"
543 #include "vstream.h"
544
545 /* Application-specific. */
546
547 /*
548 * Forward declarations.
549 */
550 static int vstream_buf_get_ready(VBUF *);
551 static int vstream_buf_put_ready(VBUF *);
552 static int vstream_buf_space(VBUF *, ssize_t);
553
554 /*
555 * Initialization of the three pre-defined streams. Pre-allocate a static
556 * I/O buffer for the standard error stream, so that the error handler can
557 * produce a diagnostic even when memory allocation fails.
558 */
559 static unsigned char vstream_fstd_buf[VSTREAM_BUFSIZE];
560
561 VSTREAM vstream_fstd[] = {
562 {{
563 0, /* flags */
564 0, 0, 0, 0, /* buffer */
565 vstream_buf_get_ready, vstream_buf_put_ready, vstream_buf_space,
566 }, STDIN_FILENO, (VSTREAM_RW_FN) timed_read, (VSTREAM_RW_FN) timed_write,
567 0,},
568 {{
569 0, /* flags */
570 0, 0, 0, 0, /* buffer */
571 vstream_buf_get_ready, vstream_buf_put_ready, vstream_buf_space,
572 }, STDOUT_FILENO, (VSTREAM_RW_FN) timed_read, (VSTREAM_RW_FN) timed_write,
573 0,},
574 {{
575 VBUF_FLAG_FIXED | VSTREAM_FLAG_WRITE,
576 vstream_fstd_buf, VSTREAM_BUFSIZE, VSTREAM_BUFSIZE, vstream_fstd_buf,
577 vstream_buf_get_ready, vstream_buf_put_ready, vstream_buf_space,
578 }, STDERR_FILENO, (VSTREAM_RW_FN) timed_read, (VSTREAM_RW_FN) timed_write,
579 VSTREAM_BUFSIZE,},
580 };
581
582 #define VSTREAM_STATIC(v) ((v) >= VSTREAM_IN && (v) <= VSTREAM_ERR)
583
584 /*
585 * A bunch of macros to make some expressions more readable. XXX We're
586 * assuming that O_RDONLY == 0, O_WRONLY == 1, O_RDWR == 2.
587 */
588 #define VSTREAM_ACC_MASK(f) ((f) & (O_APPEND | O_WRONLY | O_RDWR))
589
590 #define VSTREAM_CAN_READ(f) (VSTREAM_ACC_MASK(f) == O_RDONLY \
591 || VSTREAM_ACC_MASK(f) == O_RDWR)
592 #define VSTREAM_CAN_WRITE(f) (VSTREAM_ACC_MASK(f) & O_WRONLY \
593 || VSTREAM_ACC_MASK(f) & O_RDWR \
594 || VSTREAM_ACC_MASK(f) & O_APPEND)
595
596 #define VSTREAM_BUF_COUNT(bp, n) \
597 ((bp)->flags & VSTREAM_FLAG_READ ? -(n) : (n))
598
599 #define VSTREAM_BUF_AT_START(bp) { \
600 (bp)->cnt = VSTREAM_BUF_COUNT((bp), (bp)->len); \
601 (bp)->ptr = (bp)->data; \
602 }
603
604 #define VSTREAM_BUF_AT_OFFSET(bp, offset) { \
605 (bp)->ptr = (bp)->data + (offset); \
606 (bp)->cnt = VSTREAM_BUF_COUNT(bp, (bp)->len - (offset)); \
607 }
608
609 #define VSTREAM_BUF_AT_END(bp) { \
610 (bp)->cnt = 0; \
611 (bp)->ptr = (bp)->data + (bp)->len; \
612 }
613
614 #define VSTREAM_BUF_ZERO(bp) { \
615 (bp)->flags = 0; \
616 (bp)->data = (bp)->ptr = 0; \
617 (bp)->len = (bp)->cnt = 0; \
618 }
619
620 #define VSTREAM_BUF_ACTIONS(bp, get_action, put_action, space_action) { \
621 (bp)->get_ready = (get_action); \
622 (bp)->put_ready = (put_action); \
623 (bp)->space = (space_action); \
624 }
625
626 #define VSTREAM_SAVE_STATE(stream, buffer, filedes) { \
627 stream->buffer = stream->buf; \
628 stream->filedes = stream->fd; \
629 }
630
631 #define VSTREAM_RESTORE_STATE(stream, buffer, filedes) do { \
632 stream->buffer.flags = stream->buf.flags; \
633 stream->buf = stream->buffer; \
634 stream->fd = stream->filedes; \
635 } while(0)
636
637 #define VSTREAM_FORK_STATE(stream, buffer, filedes) { \
638 stream->buffer = stream->buf; \
639 stream->filedes = stream->fd; \
640 stream->buffer.data = stream->buffer.ptr = 0; \
641 stream->buffer.len = stream->buffer.cnt = 0; \
642 stream->buffer.flags &= ~VSTREAM_FLAG_FIXED; \
643 };
644
645 #define VSTREAM_FLAG_READ_DOUBLE (VSTREAM_FLAG_READ | VSTREAM_FLAG_DOUBLE)
646 #define VSTREAM_FLAG_WRITE_DOUBLE (VSTREAM_FLAG_WRITE | VSTREAM_FLAG_DOUBLE)
647
648 #define VSTREAM_FFLUSH_SOME(stream) \
649 vstream_fflush_some((stream), (stream)->buf.len - (stream)->buf.cnt)
650
651 /* Note: this does not change a negative result into a zero result. */
652 #define VSTREAM_SUB_TIME(x, y, z) \
653 do { \
654 (x).tv_sec = (y).tv_sec - (z).tv_sec; \
655 (x).tv_usec = (y).tv_usec - (z).tv_usec; \
656 while ((x).tv_usec < 0) { \
657 (x).tv_usec += 1000000; \
658 (x).tv_sec -= 1; \
659 } \
660 while ((x).tv_usec >= 1000000) { \
661 (x).tv_usec -= 1000000; \
662 (x).tv_sec += 1; \
663 } \
664 } while (0)
665
666 #define VSTREAM_ADD_TIME(x, y, z) \
667 do { \
668 (x).tv_sec = (y).tv_sec + (z).tv_sec; \
669 (x).tv_usec = (y).tv_usec + (z).tv_usec; \
670 while ((x).tv_usec >= 1000000) { \
671 (x).tv_usec -= 1000000; \
672 (x).tv_sec += 1; \
673 } \
674 } while (0)
675
676 /* vstream_buf_init - initialize buffer */
677
vstream_buf_init(VBUF * bp,int flags)678 static void vstream_buf_init(VBUF *bp, int flags)
679 {
680
681 /*
682 * Initialize the buffer such that the first data access triggers a
683 * buffer boundary action.
684 */
685 VSTREAM_BUF_ZERO(bp);
686 VSTREAM_BUF_ACTIONS(bp,
687 VSTREAM_CAN_READ(flags) ? vstream_buf_get_ready : 0,
688 VSTREAM_CAN_WRITE(flags) ? vstream_buf_put_ready : 0,
689 vstream_buf_space);
690 }
691
692 /* vstream_buf_alloc - allocate buffer memory */
693
vstream_buf_alloc(VBUF * bp,ssize_t len)694 static void vstream_buf_alloc(VBUF *bp, ssize_t len)
695 {
696 VSTREAM *stream = VBUF_TO_APPL(bp, VSTREAM, buf);
697 ssize_t used = bp->ptr - bp->data;
698 const char *myname = "vstream_buf_alloc";
699
700 if (len < bp->len)
701 msg_panic("%s: attempt to shrink buffer", myname);
702 if (bp->flags & VSTREAM_FLAG_FIXED)
703 msg_panic("%s: unable to extend fixed-size buffer", myname);
704
705 /*
706 * Late buffer allocation allows the user to override the default policy.
707 * If a buffer already exists, allow for the presence of (output) data.
708 */
709 bp->data = (unsigned char *)
710 (bp->data ? myrealloc((void *) bp->data, len) : mymalloc(len));
711 if (bp->flags & VSTREAM_FLAG_MEMORY)
712 memset(bp->data + bp->len, 0, len - bp->len);
713 bp->len = len;
714 if (bp->flags & VSTREAM_FLAG_READ) {
715 bp->ptr = bp->data + used;
716 if (bp->flags & VSTREAM_FLAG_DOUBLE)
717 VSTREAM_SAVE_STATE(stream, read_buf, read_fd);
718 } else {
719 VSTREAM_BUF_AT_OFFSET(bp, used);
720 if (bp->flags & VSTREAM_FLAG_DOUBLE)
721 VSTREAM_SAVE_STATE(stream, write_buf, write_fd);
722 }
723 }
724
725 /* vstream_buf_wipe - reset buffer to initial state */
726
vstream_buf_wipe(VBUF * bp)727 static void vstream_buf_wipe(VBUF *bp)
728 {
729 if ((bp->flags & VBUF_FLAG_FIXED) == 0 && bp->data)
730 myfree((void *) bp->data);
731 VSTREAM_BUF_ZERO(bp);
732 VSTREAM_BUF_ACTIONS(bp, 0, 0, 0);
733 }
734
735 /* vstream_fflush_some - flush some buffered data */
736
vstream_fflush_some(VSTREAM * stream,ssize_t to_flush)737 static int vstream_fflush_some(VSTREAM *stream, ssize_t to_flush)
738 {
739 const char *myname = "vstream_fflush_some";
740 VBUF *bp = &stream->buf;
741 ssize_t used;
742 ssize_t left_over;
743 void *data;
744 ssize_t len;
745 ssize_t n;
746 int timeout;
747 struct timeval before;
748 struct timeval elapsed;
749 struct timeval bonus;
750
751 /*
752 * Sanity checks. It is illegal to flush a read-only stream. Otherwise,
753 * if there is buffered input, discard the input. If there is buffered
754 * output, require that the amount to flush is larger than the amount to
755 * keep, so that we can memcpy() the residue.
756 */
757 if (bp->put_ready == 0)
758 msg_panic("%s: read-only stream", myname);
759 switch (bp->flags & (VSTREAM_FLAG_WRITE | VSTREAM_FLAG_READ)) {
760 case VSTREAM_FLAG_READ: /* discard input */
761 VSTREAM_BUF_AT_END(bp);
762 /* FALLTHROUGH */
763 case 0: /* flush after seek? */
764 return ((bp->flags & VSTREAM_FLAG_ERR) ? VSTREAM_EOF : 0);
765 case VSTREAM_FLAG_WRITE: /* output buffered */
766 break;
767 case VSTREAM_FLAG_WRITE | VSTREAM_FLAG_READ:
768 msg_panic("%s: read/write stream", myname);
769 }
770 used = bp->len - bp->cnt;
771 left_over = used - to_flush;
772
773 if (msg_verbose > 2 && stream != VSTREAM_ERR)
774 msg_info("%s: fd %d flush %ld", myname, stream->fd, (long) to_flush);
775 if (to_flush < 0 || left_over < 0)
776 msg_panic("%s: bad to_flush %ld", myname, (long) to_flush);
777 if (to_flush < left_over)
778 msg_panic("%s: to_flush < left_over", myname);
779 if (to_flush == 0)
780 return ((bp->flags & VSTREAM_FLAG_ERR) ? VSTREAM_EOF : 0);
781 if (bp->flags & VSTREAM_FLAG_ERR)
782 return (VSTREAM_EOF);
783
784 /*
785 * When flushing a buffer, allow for partial writes. These can happen
786 * while talking to a network. Update the cached file seek position, if
787 * any.
788 *
789 * When deadlines are enabled, we count the elapsed time for each write
790 * operation instead of simply comparing the time-of-day clock with a
791 * per-stream deadline. The latter could result in anomalies when an
792 * application does lengthy processing between write operations. Keep in
793 * mind that a receiver may not be able to keep up when a sender suddenly
794 * floods it with a lot of data as it tries to catch up with a deadline.
795 */
796 for (data = (void *) bp->data, len = to_flush; len > 0; len -= n, data += n) {
797 if (bp->flags & VSTREAM_FLAG_DEADLINE) {
798 timeout = stream->time_limit.tv_sec + (stream->time_limit.tv_usec > 0);
799 if (timeout <= 0) {
800 bp->flags |= (VSTREAM_FLAG_WR_ERR | VSTREAM_FLAG_WR_TIMEOUT);
801 errno = ETIMEDOUT;
802 return (VSTREAM_EOF);
803 }
804 if (len == to_flush)
805 GETTIMEOFDAY(&before);
806 else
807 before = stream->iotime;
808 } else
809 timeout = stream->timeout;
810 if ((n = stream->write_fn(stream->fd, data, len, timeout, stream->context)) <= 0) {
811 bp->flags |= VSTREAM_FLAG_WR_ERR;
812 if (errno == ETIMEDOUT) {
813 bp->flags |= VSTREAM_FLAG_WR_TIMEOUT;
814 stream->time_limit.tv_sec = stream->time_limit.tv_usec = 0;
815 }
816 return (VSTREAM_EOF);
817 }
818 if (timeout) {
819 GETTIMEOFDAY(&stream->iotime);
820 if (bp->flags & VSTREAM_FLAG_DEADLINE) {
821 VSTREAM_SUB_TIME(elapsed, stream->iotime, before);
822 VSTREAM_SUB_TIME(stream->time_limit, stream->time_limit, elapsed);
823 if (stream->min_data_rate > 0) {
824 bonus.tv_sec = n / stream->min_data_rate;
825 bonus.tv_usec = (n % stream->min_data_rate) * 1000000;
826 bonus.tv_usec /= stream->min_data_rate;
827 VSTREAM_ADD_TIME(stream->time_limit, stream->time_limit,
828 bonus);
829 if (stream->time_limit.tv_sec >= stream->timeout) {
830 stream->time_limit.tv_sec = stream->timeout;
831 stream->time_limit.tv_usec = 0;
832 }
833 }
834 }
835 }
836 if (msg_verbose > 2 && stream != VSTREAM_ERR && n != to_flush)
837 msg_info("%s: %d flushed %ld/%ld", myname, stream->fd,
838 (long) n, (long) to_flush);
839 }
840 if (bp->flags & VSTREAM_FLAG_SEEK)
841 stream->offset += to_flush;
842
843 /*
844 * Allow for partial buffer flush requests. We use memcpy() for reasons
845 * of portability to pre-ANSI environments (SunOS 4.x or Ultrix 4.x :-).
846 * This is OK because we have already verified that the to_flush count is
847 * larger than the left_over count.
848 */
849 if (left_over > 0)
850 memcpy(bp->data, bp->data + to_flush, left_over);
851 bp->cnt += to_flush;
852 bp->ptr -= to_flush;
853 return ((bp->flags & VSTREAM_FLAG_ERR) ? VSTREAM_EOF : 0);
854 }
855
856 /* vstream_fflush_delayed - delayed stream flush for double-buffered stream */
857
vstream_fflush_delayed(VSTREAM * stream)858 static int vstream_fflush_delayed(VSTREAM *stream)
859 {
860 int status;
861
862 /*
863 * Sanity check.
864 */
865 if ((stream->buf.flags & VSTREAM_FLAG_READ_DOUBLE) != VSTREAM_FLAG_READ_DOUBLE)
866 msg_panic("vstream_fflush_delayed: bad flags");
867
868 /*
869 * Temporarily swap buffers and flush unwritten data. This may seem like
870 * a lot of work, but it's peanuts compared to the write(2) call that we
871 * already have avoided. For example, delayed flush is never used on a
872 * non-pipelined SMTP connection.
873 */
874 stream->buf.flags &= ~VSTREAM_FLAG_READ;
875 VSTREAM_SAVE_STATE(stream, read_buf, read_fd);
876 stream->buf.flags |= VSTREAM_FLAG_WRITE;
877 VSTREAM_RESTORE_STATE(stream, write_buf, write_fd);
878
879 status = VSTREAM_FFLUSH_SOME(stream);
880
881 stream->buf.flags &= ~VSTREAM_FLAG_WRITE;
882 VSTREAM_SAVE_STATE(stream, write_buf, write_fd);
883 stream->buf.flags |= VSTREAM_FLAG_READ;
884 VSTREAM_RESTORE_STATE(stream, read_buf, read_fd);
885
886 return (status);
887 }
888
889 /* vstream_buf_get_ready - vbuf callback to make buffer ready for reading */
890
vstream_buf_get_ready(VBUF * bp)891 static int vstream_buf_get_ready(VBUF *bp)
892 {
893 VSTREAM *stream = VBUF_TO_APPL(bp, VSTREAM, buf);
894 const char *myname = "vstream_buf_get_ready";
895 ssize_t n;
896 struct timeval before;
897 struct timeval elapsed;
898 struct timeval bonus;
899 int timeout;
900
901 /*
902 * Detect a change of I/O direction or position. If so, flush any
903 * unwritten output immediately when the stream is single-buffered, or
904 * when the stream is double-buffered and the read buffer is empty.
905 */
906 switch (bp->flags & (VSTREAM_FLAG_WRITE | VSTREAM_FLAG_READ)) {
907 case VSTREAM_FLAG_WRITE: /* change direction */
908 if (bp->ptr > bp->data)
909 if ((bp->flags & VSTREAM_FLAG_DOUBLE) == 0
910 || stream->read_buf.cnt >= 0)
911 if (VSTREAM_FFLUSH_SOME(stream))
912 return (VSTREAM_EOF);
913 bp->flags &= ~VSTREAM_FLAG_WRITE;
914 if (bp->flags & VSTREAM_FLAG_DOUBLE)
915 VSTREAM_SAVE_STATE(stream, write_buf, write_fd);
916 /* FALLTHROUGH */
917 case 0: /* change position */
918 bp->flags |= VSTREAM_FLAG_READ;
919 if (bp->flags & VSTREAM_FLAG_DOUBLE) {
920 VSTREAM_RESTORE_STATE(stream, read_buf, read_fd);
921 if (bp->cnt < 0)
922 return (0);
923 }
924 /* FALLTHROUGH */
925 case VSTREAM_FLAG_READ: /* no change */
926 break;
927 case VSTREAM_FLAG_WRITE | VSTREAM_FLAG_READ:
928 msg_panic("%s: read/write stream", myname);
929 }
930
931 /*
932 * If this is the first GET operation, allocate a buffer. Late buffer
933 * allocation gives the application a chance to override the default
934 * buffering policy.
935 *
936 * XXX Subtle code to set the preferred buffer size as late as possible.
937 */
938 if (stream->req_bufsize == 0)
939 stream->req_bufsize = VSTREAM_BUFSIZE;
940 if (bp->len < stream->req_bufsize)
941 vstream_buf_alloc(bp, stream->req_bufsize);
942
943 /*
944 * If the stream is double-buffered and the write buffer is not empty,
945 * this is the time to flush the write buffer. Delayed flushes reduce
946 * system call overhead, and on TCP sockets, avoid triggering Nagle's
947 * algorithm.
948 */
949 if ((bp->flags & VSTREAM_FLAG_DOUBLE)
950 && stream->write_buf.len > stream->write_buf.cnt)
951 if (vstream_fflush_delayed(stream))
952 return (VSTREAM_EOF);
953
954 /*
955 * Did we receive an EOF indication?
956 */
957 if (bp->flags & VSTREAM_FLAG_EOF)
958 return (VSTREAM_EOF);
959
960 /*
961 * Fill the buffer with as much data as we can handle, or with as much
962 * data as is available right now, whichever is less. Update the cached
963 * file seek position, if any.
964 *
965 * When deadlines are enabled, we count the elapsed time for each read
966 * operation instead of simply comparing the time-of-day clock with a
967 * per-stream deadline. The latter could result in anomalies when an
968 * application does lengthy processing between read operations. Keep in
969 * mind that a sender may get blocked, and may not be able to keep up
970 * when a receiver suddenly wants to read a lot of data as it tries to
971 * catch up with a deadline.
972 */
973 if (bp->flags & VSTREAM_FLAG_DEADLINE) {
974 timeout = stream->time_limit.tv_sec + (stream->time_limit.tv_usec > 0);
975 if (timeout <= 0) {
976 bp->flags |= (VSTREAM_FLAG_RD_ERR | VSTREAM_FLAG_RD_TIMEOUT);
977 errno = ETIMEDOUT;
978 return (VSTREAM_EOF);
979 }
980 GETTIMEOFDAY(&before);
981 } else
982 timeout = stream->timeout;
983 switch (n = stream->read_fn(stream->fd, bp->data, bp->len, timeout, stream->context)) {
984 case -1:
985 bp->flags |= VSTREAM_FLAG_RD_ERR;
986 if (errno == ETIMEDOUT) {
987 bp->flags |= VSTREAM_FLAG_RD_TIMEOUT;
988 stream->time_limit.tv_sec = stream->time_limit.tv_usec = 0;
989 }
990 return (VSTREAM_EOF);
991 case 0:
992 bp->flags |= VSTREAM_FLAG_EOF;
993 return (VSTREAM_EOF);
994 default:
995 if (timeout) {
996 GETTIMEOFDAY(&stream->iotime);
997 if (bp->flags & VSTREAM_FLAG_DEADLINE) {
998 VSTREAM_SUB_TIME(elapsed, stream->iotime, before);
999 VSTREAM_SUB_TIME(stream->time_limit, stream->time_limit, elapsed);
1000 if (stream->min_data_rate > 0) {
1001 bonus.tv_sec = n / stream->min_data_rate;
1002 bonus.tv_usec = (n % stream->min_data_rate) * 1000000;
1003 bonus.tv_usec /= stream->min_data_rate;
1004 VSTREAM_ADD_TIME(stream->time_limit, stream->time_limit,
1005 bonus);
1006 if (stream->time_limit.tv_sec >= stream->timeout) {
1007 stream->time_limit.tv_sec = stream->timeout;
1008 stream->time_limit.tv_usec = 0;
1009 }
1010 }
1011 }
1012 }
1013 if (msg_verbose > 2)
1014 msg_info("%s: fd %d got %ld", myname, stream->fd, (long) n);
1015 bp->cnt = -n;
1016 bp->ptr = bp->data;
1017 if (bp->flags & VSTREAM_FLAG_SEEK)
1018 stream->offset += n;
1019 return (0);
1020 }
1021 }
1022
1023 /* vstream_buf_put_ready - vbuf callback to make buffer ready for writing */
1024
vstream_buf_put_ready(VBUF * bp)1025 static int vstream_buf_put_ready(VBUF *bp)
1026 {
1027 VSTREAM *stream = VBUF_TO_APPL(bp, VSTREAM, buf);
1028 const char *myname = "vstream_buf_put_ready";
1029
1030 /*
1031 * Sanity checks. Detect a change of I/O direction or position. If so,
1032 * discard unread input, and reset the buffer to the beginning.
1033 */
1034 switch (bp->flags & (VSTREAM_FLAG_WRITE | VSTREAM_FLAG_READ)) {
1035 case VSTREAM_FLAG_READ: /* change direction */
1036 bp->flags &= ~VSTREAM_FLAG_READ;
1037 if (bp->flags & VSTREAM_FLAG_DOUBLE)
1038 VSTREAM_SAVE_STATE(stream, read_buf, read_fd);
1039 /* FALLTHROUGH */
1040 case 0: /* change position */
1041 bp->flags |= VSTREAM_FLAG_WRITE;
1042 if (bp->flags & VSTREAM_FLAG_DOUBLE)
1043 VSTREAM_RESTORE_STATE(stream, write_buf, write_fd);
1044 else
1045 VSTREAM_BUF_AT_START(bp);
1046 /* FALLTHROUGH */
1047 case VSTREAM_FLAG_WRITE: /* no change */
1048 break;
1049 case VSTREAM_FLAG_WRITE | VSTREAM_FLAG_READ:
1050 msg_panic("%s: read/write stream", myname);
1051 }
1052
1053 /*
1054 * Remember the direction. If this is the first PUT operation for this
1055 * stream or if the buffer is smaller than the requested size, allocate a
1056 * new buffer; obviously there is no data to be flushed yet. Otherwise,
1057 * flush the buffer.
1058 *
1059 * XXX Subtle code to set the preferred buffer size as late as possible.
1060 */
1061 if (stream->req_bufsize == 0)
1062 stream->req_bufsize = VSTREAM_BUFSIZE;
1063 if (bp->len < stream->req_bufsize) {
1064 vstream_buf_alloc(bp, stream->req_bufsize);
1065 } else if (bp->cnt <= 0) {
1066 if (VSTREAM_FFLUSH_SOME(stream))
1067 return (VSTREAM_EOF);
1068 }
1069 return (0);
1070 }
1071
1072 /* vstream_buf_space - reserve space ahead of time */
1073
vstream_buf_space(VBUF * bp,ssize_t want)1074 static int vstream_buf_space(VBUF *bp, ssize_t want)
1075 {
1076 VSTREAM *stream = VBUF_TO_APPL(bp, VSTREAM, buf);
1077 ssize_t used;
1078 ssize_t incr;
1079 ssize_t shortage;
1080 const char *myname = "vstream_buf_space";
1081
1082 /*
1083 * Sanity checks. Reserving space implies writing. It is illegal to write
1084 * to a read-only stream. Detect a change of I/O direction or position.
1085 * If so, reset the buffer to the beginning.
1086 */
1087 if (bp->put_ready == 0)
1088 msg_panic("%s: read-only stream", myname);
1089 if (want < 0)
1090 msg_panic("%s: bad length %ld", myname, (long) want);
1091 switch (bp->flags & (VSTREAM_FLAG_READ | VSTREAM_FLAG_WRITE)) {
1092 case VSTREAM_FLAG_READ: /* change direction */
1093 bp->flags &= ~VSTREAM_FLAG_READ;
1094 if (bp->flags & VSTREAM_FLAG_DOUBLE)
1095 VSTREAM_SAVE_STATE(stream, read_buf, read_fd);
1096 /* FALLTHROUGH */
1097 case 0: /* change position */
1098 bp->flags |= VSTREAM_FLAG_WRITE;
1099 if (bp->flags & VSTREAM_FLAG_DOUBLE)
1100 VSTREAM_RESTORE_STATE(stream, write_buf, write_fd);
1101 else
1102 VSTREAM_BUF_AT_START(bp);
1103 /* FALLTHROUGH */
1104 case VSTREAM_FLAG_WRITE: /* no change */
1105 break;
1106 case VSTREAM_FLAG_READ | VSTREAM_FLAG_WRITE:
1107 msg_panic("%s: read/write stream", myname);
1108 }
1109
1110 /*
1111 * See if enough space is available. If not, flush a multiple of
1112 * VSTREAM_BUFSIZE bytes and resize the buffer to a multiple of
1113 * VSTREAM_BUFSIZE. We flush multiples of VSTREAM_BUFSIZE in an attempt
1114 * to keep file updates block-aligned for better performance.
1115 *
1116 * XXX Subtle code to set the preferred buffer size as late as possible.
1117 */
1118 #define VSTREAM_TRUNCATE(count, base) (((count) / (base)) * (base))
1119 #define VSTREAM_ROUNDUP(count, base) VSTREAM_TRUNCATE(count + base - 1, base)
1120
1121 if (stream->req_bufsize == 0)
1122 stream->req_bufsize = VSTREAM_BUFSIZE;
1123 if (want > bp->cnt) {
1124 if ((used = bp->len - bp->cnt) > stream->req_bufsize)
1125 if (vstream_fflush_some(stream, VSTREAM_TRUNCATE(used, stream->req_bufsize)))
1126 return (VSTREAM_EOF);
1127 if ((shortage = (want - bp->cnt)) > 0) {
1128 if ((bp->flags & VSTREAM_FLAG_FIXED)
1129 || shortage > __MAXINT__(ssize_t) -bp->len - stream->req_bufsize) {
1130 bp->flags |= VSTREAM_FLAG_WR_ERR;
1131 } else {
1132 incr = VSTREAM_ROUNDUP(shortage, stream->req_bufsize);
1133 vstream_buf_alloc(bp, bp->len + incr);
1134 }
1135 }
1136 }
1137 return (vstream_ferror(stream) ? VSTREAM_EOF : 0); /* mmap() may fail */
1138 }
1139
1140 /* vstream_fpurge - discard unread or unwritten content */
1141
vstream_fpurge(VSTREAM * stream,int direction)1142 int vstream_fpurge(VSTREAM *stream, int direction)
1143 {
1144 const char *myname = "vstream_fpurge";
1145 VBUF *bp = &stream->buf;
1146
1147 #define VSTREAM_MAYBE_PURGE_WRITE(d, b) if ((d) & VSTREAM_PURGE_WRITE) \
1148 VSTREAM_BUF_AT_START((b))
1149 #define VSTREAM_MAYBE_PURGE_READ(d, b) if ((d) & VSTREAM_PURGE_READ) \
1150 VSTREAM_BUF_AT_END((b))
1151
1152 /*
1153 * To discard all unread contents, position the read buffer at its end,
1154 * so that we skip over any unread data, and so that the next read
1155 * operation will refill the buffer.
1156 *
1157 * To discard all unwritten content, position the write buffer at its
1158 * beginning, so that the next write operation clobbers any unwritten
1159 * data.
1160 */
1161 switch (bp->flags & (VSTREAM_FLAG_READ_DOUBLE | VSTREAM_FLAG_WRITE)) {
1162 case VSTREAM_FLAG_READ_DOUBLE:
1163 VSTREAM_MAYBE_PURGE_WRITE(direction, &stream->write_buf);
1164 /* FALLTHROUGH */
1165 case VSTREAM_FLAG_READ:
1166 VSTREAM_MAYBE_PURGE_READ(direction, bp);
1167 break;
1168 case VSTREAM_FLAG_DOUBLE:
1169 VSTREAM_MAYBE_PURGE_WRITE(direction, &stream->write_buf);
1170 VSTREAM_MAYBE_PURGE_READ(direction, &stream->read_buf);
1171 break;
1172 case VSTREAM_FLAG_WRITE_DOUBLE:
1173 VSTREAM_MAYBE_PURGE_READ(direction, &stream->read_buf);
1174 /* FALLTHROUGH */
1175 case VSTREAM_FLAG_WRITE:
1176 VSTREAM_MAYBE_PURGE_WRITE(direction, bp);
1177 break;
1178 case VSTREAM_FLAG_READ_DOUBLE | VSTREAM_FLAG_WRITE:
1179 case VSTREAM_FLAG_READ | VSTREAM_FLAG_WRITE:
1180 msg_panic("%s: read/write stream", myname);
1181 }
1182
1183 /*
1184 * Invalidate the cached file seek position.
1185 */
1186 bp->flags &= ~VSTREAM_FLAG_SEEK;
1187 stream->offset = 0;
1188
1189 return (0);
1190 }
1191
1192 /* vstream_fseek - change I/O position */
1193
vstream_fseek(VSTREAM * stream,off_t offset,int whence)1194 off_t vstream_fseek(VSTREAM *stream, off_t offset, int whence)
1195 {
1196 const char *myname = "vstream_fseek";
1197 VBUF *bp = &stream->buf;
1198
1199 /*
1200 * TODO: support data length (data length != buffer length). Without data
1201 * length information, Without explicit data length information,
1202 * vstream_memopen(O_RDONLY) has to set the VSTREAM buffer length to the
1203 * vstring payload length to avoid accessing unwritten data after
1204 * vstream_fseek(), because for lseek() compatibility, vstream_fseek()
1205 * must allow seeking past the end of a file.
1206 */
1207 if (stream->buf.flags & VSTREAM_FLAG_MEMORY) {
1208 if (whence == SEEK_CUR)
1209 offset += (bp->ptr - bp->data);
1210 else if (whence == SEEK_END)
1211 offset += bp->len;
1212 if (offset < 0) {
1213 errno = EINVAL;
1214 return (-1);
1215 }
1216 if (offset > bp->len && (bp->flags & VSTREAM_FLAG_WRITE))
1217 vstream_buf_space(bp, offset - bp->len);
1218 VSTREAM_BUF_AT_OFFSET(bp, offset);
1219 return (offset);
1220 }
1221
1222 /*
1223 * Flush any unwritten output. Discard any unread input. Position the
1224 * buffer at the end, so that the next GET or PUT operation triggers a
1225 * buffer boundary action.
1226 */
1227 switch (bp->flags & (VSTREAM_FLAG_READ | VSTREAM_FLAG_WRITE)) {
1228 case VSTREAM_FLAG_WRITE:
1229 if (bp->ptr > bp->data) {
1230 if (whence == SEEK_CUR)
1231 offset += (bp->ptr - bp->data); /* add unwritten data */
1232 else if (whence == SEEK_END)
1233 bp->flags &= ~VSTREAM_FLAG_SEEK;
1234 if (VSTREAM_FFLUSH_SOME(stream))
1235 return (-1);
1236 }
1237 VSTREAM_BUF_AT_END(bp);
1238 break;
1239 case VSTREAM_FLAG_READ:
1240 if (whence == SEEK_CUR)
1241 offset += bp->cnt; /* subtract unread data */
1242 else if (whence == SEEK_END)
1243 bp->flags &= ~VSTREAM_FLAG_SEEK;
1244 /* FALLTHROUGH */
1245 case 0:
1246 VSTREAM_BUF_AT_END(bp);
1247 break;
1248 case VSTREAM_FLAG_READ | VSTREAM_FLAG_WRITE:
1249 msg_panic("%s: read/write stream", myname);
1250 }
1251
1252 /*
1253 * Clear the read/write flags to inform the buffer boundary action
1254 * routines that we may have changed I/O position.
1255 */
1256 bp->flags &= ~(VSTREAM_FLAG_READ | VSTREAM_FLAG_WRITE);
1257
1258 /*
1259 * Shave an unnecessary system call.
1260 */
1261 if (bp->flags & VSTREAM_FLAG_NSEEK) {
1262 errno = ESPIPE;
1263 return (-1);
1264 }
1265
1266 /*
1267 * Update the cached file seek position.
1268 */
1269 if ((stream->offset = lseek(stream->fd, offset, whence)) < 0) {
1270 if (errno == ESPIPE)
1271 bp->flags |= VSTREAM_FLAG_NSEEK;
1272 } else {
1273 bp->flags |= VSTREAM_FLAG_SEEK;
1274 }
1275 bp->flags &= ~VSTREAM_FLAG_EOF;
1276 return (stream->offset);
1277 }
1278
1279 /* vstream_ftell - return file offset */
1280
vstream_ftell(VSTREAM * stream)1281 off_t vstream_ftell(VSTREAM *stream)
1282 {
1283 VBUF *bp = &stream->buf;
1284
1285 /*
1286 * Special case for memory buffer.
1287 */
1288 if (stream->buf.flags & VSTREAM_FLAG_MEMORY)
1289 return (bp->ptr - bp->data);
1290
1291 /*
1292 * Shave an unnecessary syscall.
1293 */
1294 if (bp->flags & VSTREAM_FLAG_NSEEK) {
1295 errno = ESPIPE;
1296 return (-1);
1297 }
1298
1299 /*
1300 * Use the cached file offset when available. This is the offset after
1301 * the last read, write or seek operation.
1302 */
1303 if ((bp->flags & VSTREAM_FLAG_SEEK) == 0) {
1304 if ((stream->offset = lseek(stream->fd, (off_t) 0, SEEK_CUR)) < 0) {
1305 bp->flags |= VSTREAM_FLAG_NSEEK;
1306 return (-1);
1307 }
1308 bp->flags |= VSTREAM_FLAG_SEEK;
1309 }
1310
1311 /*
1312 * If this is a read buffer, subtract the number of unread bytes from the
1313 * cached offset. Remember that read counts are negative.
1314 */
1315 if (bp->flags & VSTREAM_FLAG_READ)
1316 return (stream->offset + bp->cnt);
1317
1318 /*
1319 * If this is a write buffer, add the number of unwritten bytes to the
1320 * cached offset.
1321 */
1322 if (bp->flags & VSTREAM_FLAG_WRITE)
1323 return (stream->offset + (bp->ptr - bp->data));
1324
1325 /*
1326 * Apparently, this is a new buffer, or a buffer after seek, so there is
1327 * no need to account for unread or unwritten data.
1328 */
1329 return (stream->offset);
1330 }
1331
1332 /* vstream_subopen - initialize everything except buffers and I/O handlers */
1333
vstream_subopen(void)1334 static VSTREAM *vstream_subopen(void)
1335 {
1336 VSTREAM *stream;
1337
1338 /* Note: memset() is not a portable way to initialize non-integer types. */
1339 stream = (VSTREAM *) mymalloc(sizeof(*stream));
1340 stream->offset = 0;
1341 stream->path = 0;
1342 stream->pid = 0;
1343 stream->waitpid_fn = 0;
1344 stream->timeout = 0;
1345 stream->context = 0;
1346 stream->jbuf = 0;
1347 stream->iotime.tv_sec = stream->iotime.tv_usec = 0;
1348 stream->time_limit.tv_sec = stream->time_limit.tv_usec = 0;
1349 stream->req_bufsize = 0;
1350 stream->vstring = 0;
1351 stream->min_data_rate = 0;
1352 return (stream);
1353 }
1354
1355 /* vstream_fdopen - add buffering to pre-opened stream */
1356
vstream_fdopen(int fd,int flags)1357 VSTREAM *vstream_fdopen(int fd, int flags)
1358 {
1359 VSTREAM *stream;
1360
1361 /*
1362 * Sanity check.
1363 */
1364 if (fd < 0)
1365 msg_panic("vstream_fdopen: bad file %d", fd);
1366
1367 /*
1368 * Initialize buffers etc. but do as little as possible. Late buffer
1369 * allocation etc. gives the application a chance to override default
1370 * policies. Either this, or the vstream*open() routines would have to
1371 * have a really ugly interface with lots of mostly-unused arguments (can
1372 * you say VMS?).
1373 */
1374 stream = vstream_subopen();
1375 stream->fd = fd;
1376 stream->read_fn = VSTREAM_CAN_READ(flags) ? (VSTREAM_RW_FN) timed_read : 0;
1377 stream->write_fn = VSTREAM_CAN_WRITE(flags) ? (VSTREAM_RW_FN) timed_write : 0;
1378 vstream_buf_init(&stream->buf, flags);
1379 return (stream);
1380 }
1381
1382 /* vstream_fopen - open buffered file stream */
1383
vstream_fopen(const char * path,int flags,mode_t mode)1384 VSTREAM *vstream_fopen(const char *path, int flags, mode_t mode)
1385 {
1386 VSTREAM *stream;
1387 int fd;
1388
1389 if ((fd = open(path, flags, mode)) < 0) {
1390 return (0);
1391 } else {
1392 stream = vstream_fdopen(fd, flags);
1393 stream->path = mystrdup(path);
1394 return (stream);
1395 }
1396 }
1397
1398 /* vstream_fflush - flush write buffer */
1399
vstream_fflush(VSTREAM * stream)1400 int vstream_fflush(VSTREAM *stream)
1401 {
1402
1403 /*
1404 * With VSTRING, the write pointer must be positioned behind the end of
1405 * data. But vstream_fseek() changes the write position, and causes the
1406 * data length to be forgotten. Before flushing to vstream, remember the
1407 * current write position, move the write pointer and do what needs to be
1408 * done, then move the write pointer back to the saved location.
1409 */
1410 if (stream->buf.flags & VSTREAM_FLAG_MEMORY) {
1411 if (stream->buf.flags & VSTREAM_FLAG_WRITE) {
1412 VSTRING *string = stream->vstring;
1413
1414 #ifdef PENDING_VSTREAM_FSEEK_FOR_MEMORY
1415 VSTREAM_BUF_AT_OFFSET(&stream->buf, stream->buf.data_len);
1416 #endif
1417 memcpy(&string->vbuf, &stream->buf, sizeof(stream->buf));
1418 string->vbuf.flags &= VSTRING_FLAG_MASK;
1419 VSTRING_TERMINATE(string);
1420 }
1421 return (0);
1422 }
1423 if ((stream->buf.flags & VSTREAM_FLAG_READ_DOUBLE)
1424 == VSTREAM_FLAG_READ_DOUBLE
1425 && stream->write_buf.len > stream->write_buf.cnt)
1426 vstream_fflush_delayed(stream);
1427 return (VSTREAM_FFLUSH_SOME(stream));
1428 }
1429
1430 /* vstream_fclose - close buffered stream */
1431
vstream_fclose(VSTREAM * stream)1432 int vstream_fclose(VSTREAM *stream)
1433 {
1434 int err;
1435
1436 /*
1437 * NOTE: Negative file descriptors are not part of the external
1438 * interface. They are for internal use only, in order to support
1439 * vstream_fdclose() without a lot of code duplication. Applications that
1440 * rely on negative VSTREAM file descriptors will break without warning.
1441 */
1442 if (stream->pid != 0)
1443 msg_panic("vstream_fclose: stream has process");
1444 if ((stream->buf.flags & VSTREAM_FLAG_MEMORY)
1445 || ((stream->buf.flags & VSTREAM_FLAG_WRITE_DOUBLE) != 0
1446 && stream->fd >= 0))
1447 vstream_fflush(stream);
1448 /* Do not remove: vstream_fdclose() depends on this error test. */
1449 err = vstream_ferror(stream);
1450 if (stream->buf.flags & VSTREAM_FLAG_DOUBLE) {
1451 if (stream->read_fd >= 0)
1452 err |= close(stream->read_fd);
1453 if (stream->write_fd != stream->read_fd)
1454 if (stream->write_fd >= 0)
1455 err |= close(stream->write_fd);
1456 vstream_buf_wipe(&stream->read_buf);
1457 vstream_buf_wipe(&stream->write_buf);
1458 stream->buf = stream->read_buf;
1459 } else {
1460 if (stream->fd >= 0)
1461 err |= close(stream->fd);
1462 if ((stream->buf.flags & VSTREAM_FLAG_MEMORY) == 0)
1463 vstream_buf_wipe(&stream->buf);
1464 }
1465 if (stream->path)
1466 myfree(stream->path);
1467 if (stream->jbuf)
1468 myfree((void *) stream->jbuf);
1469 if (stream->vstring && (stream->buf.flags & VSTREAM_FLAG_OWN_VSTRING))
1470 vstring_free(stream->vstring);
1471 if (!VSTREAM_STATIC(stream))
1472 myfree((void *) stream);
1473 return (err ? VSTREAM_EOF : 0);
1474 }
1475
1476 /* vstream_fdclose - close stream, leave file(s) open */
1477
vstream_fdclose(VSTREAM * stream)1478 int vstream_fdclose(VSTREAM *stream)
1479 {
1480
1481 /*
1482 * Flush unwritten output, just like vstream_fclose(). Errors are
1483 * reported by vstream_fclose().
1484 */
1485 if ((stream->buf.flags & VSTREAM_FLAG_WRITE_DOUBLE) != 0)
1486 (void) vstream_fflush(stream);
1487
1488 /*
1489 * NOTE: Negative file descriptors are not part of the external
1490 * interface. They are for internal use only, in order to support
1491 * vstream_fdclose() without a lot of code duplication. Applications that
1492 * rely on negative VSTREAM file descriptors will break without warning.
1493 */
1494 if (stream->buf.flags & VSTREAM_FLAG_DOUBLE) {
1495 stream->fd = stream->read_fd = stream->write_fd = -1;
1496 } else {
1497 stream->fd = -1;
1498 }
1499 return (vstream_fclose(stream));
1500 }
1501
1502 /* vstream_printf - formatted print to stdout */
1503
vstream_printf(const char * fmt,...)1504 VSTREAM *vstream_printf(const char *fmt,...)
1505 {
1506 VSTREAM *stream = VSTREAM_OUT;
1507 va_list ap;
1508
1509 va_start(ap, fmt);
1510 vbuf_print(&stream->buf, fmt, ap);
1511 va_end(ap);
1512 return (stream);
1513 }
1514
1515 /* vstream_fprintf - formatted print to buffered stream */
1516
vstream_fprintf(VSTREAM * stream,const char * fmt,...)1517 VSTREAM *vstream_fprintf(VSTREAM *stream, const char *fmt,...)
1518 {
1519 va_list ap;
1520
1521 va_start(ap, fmt);
1522 vbuf_print(&stream->buf, fmt, ap);
1523 va_end(ap);
1524 return (stream);
1525 }
1526
1527 /* vstream_fputs - write string to stream */
1528
vstream_fputs(const char * str,VSTREAM * stream)1529 int vstream_fputs(const char *str, VSTREAM *stream)
1530 {
1531 int ch;
1532
1533 while ((ch = *str++) != 0)
1534 if (VSTREAM_PUTC(ch, stream) == VSTREAM_EOF)
1535 return (VSTREAM_EOF);
1536 return (0);
1537 }
1538
1539 /* vstream_fread_buf - unformatted read to VSTRING */
1540
vstream_fread_buf(VSTREAM * fp,VSTRING * vp,ssize_t len)1541 ssize_t vstream_fread_buf(VSTREAM *fp, VSTRING *vp, ssize_t len)
1542 {
1543 ssize_t ret;
1544
1545 VSTRING_RESET(vp);
1546 VSTRING_SPACE(vp, len);
1547 ret = vstream_fread(fp, vstring_str(vp), len);
1548 if (ret > 0)
1549 VSTRING_AT_OFFSET(vp, ret);
1550 return (ret);
1551 }
1552
1553 /* vstream_fread_app - unformatted read to VSTRING */
1554
vstream_fread_app(VSTREAM * fp,VSTRING * vp,ssize_t len)1555 ssize_t vstream_fread_app(VSTREAM *fp, VSTRING *vp, ssize_t len)
1556 {
1557 ssize_t ret;
1558
1559 VSTRING_SPACE(vp, len);
1560 ret = vstream_fread(fp, vstring_end(vp), len);
1561 if (ret > 0)
1562 VSTRING_AT_OFFSET(vp, VSTRING_LEN(vp) + ret);
1563 return (ret);
1564 }
1565
1566 /* vstream_control - fine control */
1567
vstream_control(VSTREAM * stream,int name,...)1568 void vstream_control(VSTREAM *stream, int name,...)
1569 {
1570 const char *myname = "vstream_control";
1571 va_list ap;
1572 int floor;
1573 int old_fd;
1574 ssize_t req_bufsize = 0;
1575 VSTREAM *stream2;
1576 int min_data_rate;
1577
1578 #define SWAP(type,a,b) do { type temp = (a); (a) = (b); (b) = (temp); } while (0)
1579
1580 /*
1581 * A crude 'allow' filter for memory streams.
1582 */
1583 int memory_ops =
1584 ((1 << VSTREAM_CTL_END) | (1 << VSTREAM_CTL_CONTEXT)
1585 | (1 << VSTREAM_CTL_PATH) | (1 << VSTREAM_CTL_EXCEPT)
1586 | (1 << VSTREAM_CTL_OWN_VSTRING));
1587
1588 for (va_start(ap, name); name != VSTREAM_CTL_END; name = va_arg(ap, int)) {
1589 if ((stream->buf.flags & VSTREAM_FLAG_MEMORY)
1590 && (memory_ops & (1 << name)) == 0)
1591 msg_panic("%s: memory stream does not support VSTREAM_CTL_%d",
1592 VSTREAM_PATH(stream), name);
1593 switch (name) {
1594 case VSTREAM_CTL_READ_FN:
1595 stream->read_fn = va_arg(ap, VSTREAM_RW_FN);
1596 break;
1597 case VSTREAM_CTL_WRITE_FN:
1598 stream->write_fn = va_arg(ap, VSTREAM_RW_FN);
1599 break;
1600 case VSTREAM_CTL_CONTEXT:
1601 stream->context = va_arg(ap, void *);
1602 break;
1603 case VSTREAM_CTL_PATH:
1604 if (stream->path)
1605 myfree(stream->path);
1606 stream->path = mystrdup(va_arg(ap, char *));
1607 break;
1608 case VSTREAM_CTL_DOUBLE:
1609 if ((stream->buf.flags & VSTREAM_FLAG_DOUBLE) == 0) {
1610 stream->buf.flags |= VSTREAM_FLAG_DOUBLE;
1611 if (stream->buf.flags & VSTREAM_FLAG_READ) {
1612 VSTREAM_SAVE_STATE(stream, read_buf, read_fd);
1613 VSTREAM_FORK_STATE(stream, write_buf, write_fd);
1614 } else {
1615 VSTREAM_SAVE_STATE(stream, write_buf, write_fd);
1616 VSTREAM_FORK_STATE(stream, read_buf, read_fd);
1617 }
1618 }
1619 break;
1620 case VSTREAM_CTL_READ_FD:
1621 if ((stream->buf.flags & VSTREAM_FLAG_DOUBLE) == 0)
1622 msg_panic("VSTREAM_CTL_READ_FD requires double buffering");
1623 stream->read_fd = va_arg(ap, int);
1624 stream->buf.flags |= VSTREAM_FLAG_NSEEK;
1625 break;
1626 case VSTREAM_CTL_WRITE_FD:
1627 if ((stream->buf.flags & VSTREAM_FLAG_DOUBLE) == 0)
1628 msg_panic("VSTREAM_CTL_WRITE_FD requires double buffering");
1629 stream->write_fd = va_arg(ap, int);
1630 stream->buf.flags |= VSTREAM_FLAG_NSEEK;
1631 break;
1632 case VSTREAM_CTL_SWAP_FD:
1633 stream2 = va_arg(ap, VSTREAM *);
1634 if ((stream->buf.flags & VSTREAM_FLAG_DOUBLE)
1635 != (stream2->buf.flags & VSTREAM_FLAG_DOUBLE))
1636 msg_panic("VSTREAM_CTL_SWAP_FD can't swap descriptors between "
1637 "single-buffered and double-buffered streams");
1638 if (stream->buf.flags & VSTREAM_FLAG_DOUBLE) {
1639 SWAP(int, stream->read_fd, stream2->read_fd);
1640 SWAP(int, stream->write_fd, stream2->write_fd);
1641 stream->fd = ((stream->buf.flags & VSTREAM_FLAG_WRITE) ?
1642 stream->write_fd : stream->read_fd);
1643 } else {
1644 SWAP(int, stream->fd, stream2->fd);
1645 }
1646 break;
1647 case VSTREAM_CTL_TIMEOUT:
1648 if (stream->timeout == 0)
1649 GETTIMEOFDAY(&stream->iotime);
1650 stream->timeout = va_arg(ap, int);
1651 if (stream->timeout < 0)
1652 msg_panic("%s: bad timeout %d", myname, stream->timeout);
1653 break;
1654 case VSTREAM_CTL_EXCEPT:
1655 if (stream->jbuf == 0)
1656 stream->jbuf =
1657 (VSTREAM_JMP_BUF *) mymalloc(sizeof(VSTREAM_JMP_BUF));
1658 break;
1659
1660 #ifdef VSTREAM_CTL_DUPFD
1661
1662 #define VSTREAM_TRY_DUPFD(backup, fd, floor) do { \
1663 if (((backup) = (fd)) < floor) { \
1664 if (((fd) = fcntl((backup), F_DUPFD, (floor))) < 0) \
1665 msg_fatal("fcntl F_DUPFD %d: %m", (floor)); \
1666 (void) close(backup); \
1667 } \
1668 } while (0)
1669
1670 case VSTREAM_CTL_DUPFD:
1671 floor = va_arg(ap, int);
1672 if (stream->buf.flags & VSTREAM_FLAG_DOUBLE) {
1673 VSTREAM_TRY_DUPFD(old_fd, stream->read_fd, floor);
1674 if (stream->write_fd == old_fd)
1675 stream->write_fd = stream->read_fd;
1676 else
1677 VSTREAM_TRY_DUPFD(old_fd, stream->write_fd, floor);
1678 stream->fd = (stream->buf.flags & VSTREAM_FLAG_READ) ?
1679 stream->read_fd : stream->write_fd;
1680 } else {
1681 VSTREAM_TRY_DUPFD(old_fd, stream->fd, floor);
1682 }
1683 break;
1684 #endif
1685
1686 /*
1687 * Postpone memory (re)allocation until the space is needed.
1688 */
1689 case VSTREAM_CTL_BUFSIZE:
1690 req_bufsize = va_arg(ap, ssize_t);
1691 /* Heuristic to detect missing (ssize_t) type cast on LP64 hosts. */
1692 if (req_bufsize < 0 || req_bufsize > INT_MAX)
1693 msg_panic("unreasonable VSTREAM_CTL_BUFSIZE request: %ld",
1694 (long) req_bufsize);
1695 if ((stream->buf.flags & VSTREAM_FLAG_FIXED) == 0
1696 && req_bufsize > stream->req_bufsize) {
1697 if (msg_verbose)
1698 msg_info("fd=%d: stream buffer size old=%ld new=%ld",
1699 vstream_fileno(stream),
1700 (long) stream->req_bufsize,
1701 (long) req_bufsize);
1702 stream->req_bufsize = req_bufsize;
1703 }
1704 break;
1705
1706 /*
1707 * Make no gettimeofday() etc. system call until we really know
1708 * that we need to do I/O. This avoids a performance hit when
1709 * sending or receiving body content one line at a time.
1710 */
1711 case VSTREAM_CTL_STOP_DEADLINE:
1712 stream->buf.flags &= ~VSTREAM_FLAG_DEADLINE;
1713 break;
1714 case VSTREAM_CTL_START_DEADLINE:
1715 if (stream->timeout <= 0)
1716 msg_panic("%s: bad timeout %d", myname, stream->timeout);
1717 stream->buf.flags |= VSTREAM_FLAG_DEADLINE;
1718 stream->time_limit.tv_sec = stream->timeout;
1719 stream->time_limit.tv_usec = 0;
1720 break;
1721 case VSTREAM_CTL_MIN_DATA_RATE:
1722 min_data_rate = va_arg(ap, int);
1723 if (min_data_rate < 0)
1724 msg_panic("%s: bad min_data_rate %d", myname, min_data_rate);
1725 stream->min_data_rate = min_data_rate;
1726 break;
1727 case VSTREAM_CTL_OWN_VSTRING:
1728 if ((stream->buf.flags |= VSTREAM_FLAG_MEMORY) == 0)
1729 msg_panic("%s: operation on non-VSTRING stream", myname);
1730 stream->buf.flags |= VSTREAM_FLAG_OWN_VSTRING;
1731 break;
1732 default:
1733 msg_panic("%s: bad name %d", myname, name);
1734 }
1735 }
1736 va_end(ap);
1737 }
1738
1739 /* vstream_vprintf - formatted print to stdout */
1740
vstream_vprintf(const char * format,va_list ap)1741 VSTREAM *vstream_vprintf(const char *format, va_list ap)
1742 {
1743 VSTREAM *vp = VSTREAM_OUT;
1744
1745 vbuf_print(&vp->buf, format, ap);
1746 return (vp);
1747 }
1748
1749 /* vstream_vfprintf - formatted print engine */
1750
vstream_vfprintf(VSTREAM * vp,const char * format,va_list ap)1751 VSTREAM *vstream_vfprintf(VSTREAM *vp, const char *format, va_list ap)
1752 {
1753 vbuf_print(&vp->buf, format, ap);
1754 return (vp);
1755 }
1756
1757 /* vstream_bufstat - get stream buffer status */
1758
vstream_bufstat(VSTREAM * vp,int command)1759 ssize_t vstream_bufstat(VSTREAM *vp, int command)
1760 {
1761 VBUF *bp;
1762
1763 switch (command & VSTREAM_BST_MASK_DIR) {
1764 case VSTREAM_BST_FLAG_IN:
1765 if (vp->buf.flags & VSTREAM_FLAG_READ) {
1766 bp = &vp->buf;
1767 } else if (vp->buf.flags & VSTREAM_FLAG_DOUBLE) {
1768 bp = &vp->read_buf;
1769 } else {
1770 bp = 0;
1771 }
1772 switch (command & ~VSTREAM_BST_MASK_DIR) {
1773 case VSTREAM_BST_FLAG_PEND:
1774 return (bp ? -bp->cnt : 0);
1775 /* Add other requests below. */
1776 }
1777 break;
1778 case VSTREAM_BST_FLAG_OUT:
1779 if (vp->buf.flags & VSTREAM_FLAG_WRITE) {
1780 bp = &vp->buf;
1781 } else if (vp->buf.flags & VSTREAM_FLAG_DOUBLE) {
1782 bp = &vp->write_buf;
1783 } else {
1784 bp = 0;
1785 }
1786 switch (command & ~VSTREAM_BST_MASK_DIR) {
1787 case VSTREAM_BST_FLAG_PEND:
1788 return (bp ? bp->len - bp->cnt : 0);
1789 /* Add other requests below. */
1790 }
1791 break;
1792 }
1793 msg_panic("vstream_bufstat: unknown command: %d", command);
1794 }
1795
1796 #undef vstream_peek /* API binary compatibility. */
1797
1798 /* vstream_peek - peek at a stream */
1799
vstream_peek(VSTREAM * vp)1800 ssize_t vstream_peek(VSTREAM *vp)
1801 {
1802 if (vp->buf.flags & VSTREAM_FLAG_READ) {
1803 return (-vp->buf.cnt);
1804 } else if (vp->buf.flags & VSTREAM_FLAG_DOUBLE) {
1805 return (-vp->read_buf.cnt);
1806 } else {
1807 return (0);
1808 }
1809 }
1810
1811 /* vstream_peek_data - peek at unread data */
1812
vstream_peek_data(VSTREAM * vp)1813 const char *vstream_peek_data(VSTREAM *vp)
1814 {
1815 if (vp->buf.flags & VSTREAM_FLAG_READ) {
1816 return ((const char *) vp->buf.ptr);
1817 } else if (vp->buf.flags & VSTREAM_FLAG_DOUBLE) {
1818 return ((const char *) vp->read_buf.ptr);
1819 } else {
1820 return (0);
1821 }
1822 }
1823
1824 /* vstream_memopen - open a VSTRING */
1825
vstream_memreopen(VSTREAM * stream,VSTRING * string,int flags)1826 VSTREAM *vstream_memreopen(VSTREAM *stream, VSTRING *string, int flags)
1827 {
1828 if (stream == 0)
1829 stream = vstream_subopen();
1830 else if ((stream->buf.flags & VSTREAM_FLAG_MEMORY) == 0)
1831 msg_panic("vstream_memreopen: cannot reopen non-memory stream");
1832 stream->fd = -1;
1833 stream->read_fn = 0;
1834 stream->write_fn = 0;
1835 stream->vstring = string;
1836 memcpy(&stream->buf, &stream->vstring->vbuf, sizeof(stream->buf));
1837 stream->buf.flags |= VSTREAM_FLAG_MEMORY;
1838 switch (VSTREAM_ACC_MASK(flags)) {
1839 case O_RDONLY:
1840 stream->buf.flags |= VSTREAM_FLAG_READ;
1841 /* Prevent reading unwritten data after vstream_fseek(). */
1842 stream->buf.len = stream->buf.ptr - stream->buf.data;
1843 VSTREAM_BUF_AT_OFFSET(&stream->buf, 0);
1844 break;
1845 case O_WRONLY:
1846 stream->buf.flags |= VSTREAM_FLAG_WRITE;
1847 VSTREAM_BUF_AT_OFFSET(&stream->buf, 0);
1848 break;
1849 case O_APPEND:
1850 stream->buf.flags |= VSTREAM_FLAG_WRITE;
1851 VSTREAM_BUF_AT_OFFSET(&stream->buf,
1852 stream->buf.ptr - stream->buf.data);
1853 break;
1854 default:
1855 msg_panic("vstream_memopen: flags must be one of "
1856 "O_RDONLY, O_WRONLY, or O_APPEND");
1857 }
1858 return (stream);
1859 }
1860
1861 #ifdef TEST
1862
copy_line(ssize_t bufsize)1863 static void copy_line(ssize_t bufsize)
1864 {
1865 int c;
1866
1867 /*
1868 * Demonstrates that VSTREAM_CTL_BUFSIZE increases the buffer size, but
1869 * does not decrease it. Uses VSTREAM_ERR for non-test output to avoid
1870 * interfering with the test.
1871 */
1872 vstream_fprintf(VSTREAM_ERR, "buffer size test: copy text with %ld buffer size, ignore requests to shrink\n",
1873 (long) bufsize);
1874 vstream_fflush(VSTREAM_ERR);
1875 vstream_control(VSTREAM_IN, CA_VSTREAM_CTL_BUFSIZE(bufsize), VSTREAM_CTL_END);
1876 vstream_control(VSTREAM_OUT, CA_VSTREAM_CTL_BUFSIZE(bufsize), VSTREAM_CTL_END);
1877 while ((c = VSTREAM_GETC(VSTREAM_IN)) != VSTREAM_EOF) {
1878 VSTREAM_PUTC(c, VSTREAM_OUT);
1879 if (c == '\n')
1880 break;
1881 }
1882 vstream_fflush(VSTREAM_OUT);
1883 vstream_fprintf(VSTREAM_ERR, "actual read/write buffer sizes: %ld/%ld\n\n",
1884 (long) VSTREAM_IN->buf.len, (long) VSTREAM_OUT->buf.len);
1885 vstream_fflush(VSTREAM_ERR);
1886 }
1887
printf_number(void)1888 static void printf_number(void)
1889 {
1890
1891 /*
1892 * Demonstrates that vstream_printf() use vbuf_print().
1893 */
1894 vstream_printf("formatting test: print a number\n");
1895 vstream_printf("%d\n\n", 1234567890);
1896 vstream_fflush(VSTREAM_OUT);
1897 }
1898
do_memory_stream(void)1899 static void do_memory_stream(void)
1900 {
1901 VSTRING *buf = vstring_alloc(1);
1902 VSTREAM *fp;
1903 off_t offset;
1904 int ch;
1905
1906 /*
1907 * Preload the string.
1908 */
1909 vstream_printf("memory stream test prep: prefill the VSTRING\n");
1910 vstring_strcpy(buf, "01234567");
1911 vstream_printf("VSTRING content length: %ld/%ld, content: %s\n",
1912 (long) VSTRING_LEN(buf), (long) buf->vbuf.len,
1913 vstring_str(buf));
1914 VSTREAM_PUTCHAR('\n');
1915 vstream_fflush(VSTREAM_OUT);
1916
1917 /*
1918 * Test: open the memory VSTREAM in write-only mode, and clobber it.
1919 */
1920 vstream_printf("memory stream test: open the VSTRING for writing, overwrite, close\n");
1921 fp = vstream_memopen(buf, O_WRONLY);
1922 vstream_printf("initial memory VSTREAM write offset: %ld/%ld\n",
1923 (long) vstream_ftell(fp), (long) fp->buf.len);
1924 vstream_fprintf(fp, "hallo");
1925 vstream_printf("final memory VSTREAM write offset: %ld/%ld\n",
1926 (long) vstream_ftell(fp), (long) fp->buf.len);
1927 vstream_fclose(fp);
1928 vstream_printf("VSTRING content length: %ld/%ld, content: %s\n",
1929 (long) VSTRING_LEN(buf), (long) buf->vbuf.len,
1930 vstring_str(buf));
1931 VSTREAM_PUTCHAR('\n');
1932 vstream_fflush(VSTREAM_OUT);
1933
1934 /*
1935 * Test: open the memory VSTREAM for append. vstream_memopen() sets the
1936 * buffer length to the VSTRING buffer length, and positions the write
1937 * pointer at the VSTRING write position. Write some content, then
1938 * overwrite one character.
1939 */
1940 vstream_printf("memory stream test: open the VSTRING for append, write multiple, then overwrite 1\n");
1941 fp = vstream_memopen(buf, O_APPEND);
1942 vstream_printf("initial memory VSTREAM write offset: %ld/%ld\n",
1943 (long) vstream_ftell(fp), (long) fp->buf.len);
1944 vstream_fprintf(fp, " world");
1945 vstream_printf("final memory VSTREAM write offset: %ld/%ld\n",
1946 (long) vstream_ftell(fp), (long) fp->buf.len);
1947 if (vstream_fflush(fp))
1948 msg_fatal("vstream_fflush: %m");
1949 vstream_printf("VSTRING content length: %ld/%ld, content: %s\n",
1950 (long) VSTRING_LEN(buf), (long) buf->vbuf.len,
1951 vstring_str(buf));
1952 VSTREAM_PUTCHAR('\n');
1953
1954 /*
1955 * While the stream is still open, replace the second character.
1956 */
1957 vstream_printf("replace second character and close\n");
1958 if ((offset = vstream_fseek(fp, 1, SEEK_SET)) != 1)
1959 msg_panic("unexpected vstream_fseek return: %ld, expected: %ld",
1960 (long) offset, (long) 1);
1961 VSTREAM_PUTC('e', fp);
1962
1963 /*
1964 * Skip to the end of the content, so that vstream_fflush() will update
1965 * the VSTRING with the right content length.
1966 */
1967 if ((offset = vstream_fseek(fp, VSTRING_LEN(buf), SEEK_SET)) != VSTRING_LEN(buf))
1968 msg_panic("unexpected vstream_fseek return: %ld, expected: %ld",
1969 (long) offset, (long) VSTRING_LEN(buf));
1970 vstream_fclose(fp);
1971
1972 vstream_printf("VSTRING content length: %ld/%ld, content: %s\n",
1973 (long) VSTRING_LEN(buf), (long) buf->vbuf.len,
1974 vstring_str(buf));
1975 VSTREAM_PUTCHAR('\n');
1976 vstream_fflush(VSTREAM_OUT);
1977
1978 /*
1979 * TODO: test that in write/append mode, seek past the end of data will
1980 * result in zero-filled space.
1981 */
1982
1983 /*
1984 * Test: Open the VSTRING for reading. This time, vstream_memopen() will
1985 * set the VSTREAM buffer length to the content length of the VSTRING, so
1986 * that it won't attempt to read past the end of the content.
1987 */
1988 vstream_printf("memory stream test: open VSTRING for reading, then read\n");
1989 fp = vstream_memopen(buf, O_RDONLY);
1990 vstream_printf("initial memory VSTREAM read offset: %ld/%ld\n",
1991 (long) vstream_ftell(fp), (long) fp->buf.len);
1992 vstream_printf("reading memory VSTREAM: ");
1993 while ((ch = VSTREAM_GETC(fp)) != VSTREAM_EOF)
1994 VSTREAM_PUTCHAR(ch);
1995 VSTREAM_PUTCHAR('\n');
1996 vstream_printf("final memory VSTREAM read offset: %ld/%ld\n",
1997 (long) vstream_ftell(fp), (long) fp->buf.len);
1998 vstream_printf("seeking to offset %ld should work: ",
1999 (long) fp->buf.len + 1);
2000 vstream_fflush(VSTREAM_OUT);
2001 if ((offset = vstream_fseek(fp, fp->buf.len + 1, SEEK_SET)) != fp->buf.len + 1)
2002 msg_panic("unexpected vstream_fseek return: %ld, expected: %ld",
2003 (long) offset, (long) fp->buf.len + 1);
2004 vstream_printf("PASS\n");
2005 vstream_fflush(VSTREAM_OUT);
2006 vstream_printf("VSTREAM_GETC should return VSTREAM_EOF\n");
2007 ch = VSTREAM_GETC(fp);
2008 if (ch != VSTREAM_EOF)
2009 msg_panic("unexpected vstream_fseek VSTREAM_GETC return: %d, expected: %d",
2010 ch, VSTREAM_EOF);
2011 vstream_printf("PASS\n");
2012 vstream_printf("final memory VSTREAM read offset: %ld/%ld\n",
2013 (long) vstream_ftell(fp), (long) fp->buf.len);
2014 vstream_printf("VSTRING content length: %ld/%ld, content: %s\n",
2015 (long) VSTRING_LEN(buf), (long) buf->vbuf.len,
2016 vstring_str(buf));
2017 VSTREAM_PUTCHAR('\n');
2018 vstream_fflush(VSTREAM_OUT);
2019 vstream_fclose(fp);
2020 vstring_free(buf);
2021 }
2022
2023 /*
2024 * Exercise some of the features.
2025 */
2026
2027 #include <msg_vstream.h>
2028
main(int argc,char ** argv)2029 int main(int argc, char **argv)
2030 {
2031 msg_vstream_init(argv[0], VSTREAM_ERR);
2032
2033 /*
2034 * Test buffer expansion and shrinking. Formatted print may silently
2035 * expand the write buffer and cause multiple bytes to be written.
2036 */
2037 copy_line(1); /* one-byte read/write */
2038 copy_line(2); /* two-byte read/write */
2039 copy_line(1); /* two-byte read/write */
2040 printf_number(); /* multi-byte write */
2041 do_memory_stream();
2042
2043 exit(0);
2044 }
2045
2046 #endif
2047