Skip to content

John McEleney / mailutils

Go to a project
Toggle navigation
Toggle navigation pinning
Commit 8c76d675 8c76d675333b70a0887de506073d64cd2c14643a by Sergey Poznyakoff
Options

Optimize stream seeks and I/O. Expand stream event handlers. Rewrite xscript stream. 

* include/mailutils/sys/stream.h (_MU_STR_EVENT_SET): Rename
to _MU_STR_EVENT_SET.
(_MU_STR_EVENT_CLR): Rename to _MU_STR_EVENT_CLRFLAG.
(_MU_STR_EVENT_FILLBUF, _MU_STR_EVENT_FLUSHBUF): New event codes.
(_MU_STR_EVMASK): New macro.
(_mu_stream) <cur>: Replace with pos, indicating current position
in the buffer.
(event_cb): Change signature.
(mu_stream_read_unbuffered): Remove.
(mu_stream_write_unbuffered): Remove.

* mailbox/iostream.c (mu_iostream_create): Provide the readdelim method
only if the underlying input transpor stream provides it.
* mailbox/streamref.c (mu_streamref_create_abridged): Likewise.

* mailbox/stream.c (_stream_event): New macro.
(mu_stream_read_unbuffered): Rename to static _stream_read_unbuffered.
(mu_stream_write_unbuffered): Rename to static _stream_write_unbuffered.
(_stream_advance_buffer, _stream_buffer_offset)
(_stream_orig_level): Remove macros.
(_stream_buffer_freespace): Rewrite.
(_stream_curp): New macro.
(_stream_fill_buffer): Make sure the `offset' indicates the offset
int the transport, corresponding to the beginning of the current
buffer.
(_stream_flush_buffer): Essentially rewritten.
(mu_stream_seek): Reflect changes to the _mu_stream structure. Optimize
calls to the seek method.
(_stream_skip_input_bytes): Likewise.
(mu_stream_read, _stream_scandelim, mu_stream_write): Rewrite using
new _mu_stream structure.

* mailbox/xscript-stream.c: Rewrite using stream events.

* mailbox/base64.c (_base64_encoder): Bugfix.

* libproto/pop/pop3_stream.c (_pop3_event_cb): Update signature to
match the changes above.
* examples/mimetest.c (main): Add more error checking.
* mail/testsuite/mail/write.exp: Minor fix.
1 parent fd032c6f
Inline Side-by-side
Showing 9 changed files with 202 additions and 167 deletions
...@@ -116,7 +116,7 @@ main (int argc, char **argv) ...@@ -116,7 +116,7 @@ main (int argc, char **argv)
116 MU_ASSERT (mu_mailbox_open (mbox, MU_STREAM_RDWR)); 116 MU_ASSERT (mu_mailbox_open (mbox, MU_STREAM_RDWR));
117 117
118 /* Iterate through the entire message set. */ 118 /* Iterate through the entire message set. */
119 mu_mailbox_messages_count (mbox, &count); 119 MU_ASSERT (mu_mailbox_messages_count (mbox, &count));
120 120
121 for (i = 1; i <= count; ++i) 121 for (i = 1; i <= count; ++i)
122 { 122 {
......
...@@ -24,8 +24,12 @@ ...@@ -24,8 +24,12 @@
24 24
25 #define _MU_STR_INTERN_MASK 0xf0000000 25 #define _MU_STR_INTERN_MASK 0xf0000000
26 26
27 #define _MU_STR_EVENT_SET 1 27 #define _MU_STR_EVENT_SETFLAG 0
28 #define _MU_STR_EVENT_CLR 2 28 #define _MU_STR_EVENT_CLRFLAG 1
29 #define _MU_STR_EVENT_FILLBUF 2
30 #define _MU_STR_EVENT_FLUSHBUF 3
31
32 #define _MU_STR_EVMASK(n) (1<<(n))
29 33
30 struct _mu_stream 34 struct _mu_stream
31 { 35 {
...@@ -35,7 +39,7 @@ struct _mu_stream ...@@ -35,7 +39,7 @@ struct _mu_stream
35 size_t bufsize; 39 size_t bufsize;
36 char *buffer; 40 char *buffer;
37 size_t level; 41 size_t level;
38 char *cur; 42 size_t pos;
39 43
40 int flags; 44 int flags;
41 mu_off_t offset; 45 mu_off_t offset;
...@@ -57,7 +61,7 @@ struct _mu_stream ...@@ -57,7 +61,7 @@ struct _mu_stream
57 int (*truncate) (struct _mu_stream *, mu_off_t); 61 int (*truncate) (struct _mu_stream *, mu_off_t);
58 int (*shutdown) (struct _mu_stream *, int); 62 int (*shutdown) (struct _mu_stream *, int);
59 63
60 void (*event_cb) (struct _mu_stream *, int, int); 64 void (*event_cb) (struct _mu_stream *, int code, unsigned long, void *);
61 int event_mask; 65 int event_mask;
62 66
63 const char *(*error_string) (struct _mu_stream *, int); 67 const char *(*error_string) (struct _mu_stream *, int);
...@@ -65,11 +69,6 @@ struct _mu_stream ...@@ -65,11 +69,6 @@ struct _mu_stream
65 }; 69 };
66 70
67 mu_stream_t _mu_stream_create (size_t size, int flags); 71 mu_stream_t _mu_stream_create (size_t size, int flags);
68 int mu_stream_read_unbuffered (mu_stream_t stream, void *buf, size_t size,
69 int full_read, size_t *pnread);
70 int mu_stream_write_unbuffered (mu_stream_t stream,
71 const void *buf, size_t size,
72 int full_write, size_t *pnwritten);
73 72
74 void _mu_stream_cleareof (mu_stream_t str); 73 void _mu_stream_cleareof (mu_stream_t str);
75 void _mu_stream_seteof (mu_stream_t str); 74 void _mu_stream_seteof (mu_stream_t str);
......
...@@ -41,10 +41,12 @@ struct mu_pop3_stream ...@@ -41,10 +41,12 @@ struct mu_pop3_stream
41 struct mu_buffer_query oldbuf; 41 struct mu_buffer_query oldbuf;
42 }; 42 };
43 43
44 /* Called on _MU_STR_EVENT_SETFLAG */
44 static void 45 static void
45 _pop3_event_cb (mu_stream_t str, int ev, int flags) 46 _pop3_event_cb (mu_stream_t str, int ev, unsigned long flags,
47 void *ptr MU_ARG_UNUSED)
46 { 48 {
47 if (ev == _MU_STR_EVENT_SET) 49 if (flags & _MU_STR_EOF)
48 { 50 {
49 mu_transport_t trans[2]; 51 mu_transport_t trans[2];
50 52
...@@ -78,7 +80,7 @@ mu_pop3_filter_create (mu_stream_t *pstream, mu_stream_t stream) ...@@ -78,7 +80,7 @@ mu_pop3_filter_create (mu_stream_t *pstream, mu_stream_t stream)
78 mu_stream_t str = *pstream; 80 mu_stream_t str = *pstream;
79 81
80 str->event_cb = _pop3_event_cb; 82 str->event_cb = _pop3_event_cb;
81 str->event_mask = _MU_STR_EOF; 83 str->event_mask = _MU_STR_EVMASK(_MU_STR_EVENT_SETFLAG);
82 84
83 sp->oldbuf.type = MU_TRANSPORT_OUTPUT; 85 sp->oldbuf.type = MU_TRANSPORT_OUTPUT;
84 if (mu_stream_ioctl (sp->pop3->carrier, MU_IOCTL_GET_TRANSPORT_BUFFER, 86 if (mu_stream_ioctl (sp->pop3->carrier, MU_IOCTL_GET_TRANSPORT_BUFFER,
......
...@@ -27,7 +27,7 @@ mail_test -noprompt "quit" \ ...@@ -27,7 +27,7 @@ mail_test -noprompt "quit" \
27 # Start again using the same mailbox 27 # Start again using the same mailbox
28 mail_start -reuse-spool "--file=%mbox1" 28 mail_start -reuse-spool "--file=%mbox1"
29 # Go to the last message and do delete 4 times 29 # Go to the last message and do delete 4 times
30 mail_command "4" 30 mail_command "3"
31 mail_command "delete" 31 mail_command "delete"
32 mail_command "delete" 32 mail_command "delete"
33 mail_command "delete" 33 mail_command "delete"
......
...@@ -244,6 +244,9 @@ _base64_encoder (void *xd MU_ARG_UNUSED, ...@@ -244,6 +244,9 @@ _base64_encoder (void *xd MU_ARG_UNUSED,
244 pad = 0; 244 pad = 0;
245 } 245 }
246 246
247 /* Consumed may grow bigger than isize if cmd is mu_filter_lastbuf */
248 if (consumed > iobuf->isize)
249 consumed = iobuf->isize;
247 iobuf->isize = consumed; 250 iobuf->isize = consumed;
248 iobuf->osize = nbytes; 251 iobuf->osize = nbytes;
249 return mu_filter_ok; 252 return mu_filter_ok;
......
...@@ -236,7 +236,8 @@ mu_iostream_create (mu_stream_t *pref, mu_stream_t in, mu_stream_t out) ...@@ -236,7 +236,8 @@ mu_iostream_create (mu_stream_t *pref, mu_stream_t in, mu_stream_t out)
236 return ENOMEM; 236 return ENOMEM;
237 237
238 sp->stream.read = _iostream_read; 238 sp->stream.read = _iostream_read;
239 sp->stream.readdelim = _iostream_readdelim; 239 if (in->readdelim)
240 sp->stream.readdelim = _iostream_readdelim;
240 sp->stream.write = _iostream_write; 241 sp->stream.write = _iostream_write;
241 sp->stream.flush = _iostream_flush; 242 sp->stream.flush = _iostream_flush;
242 sp->stream.open = _iostream_open; 243 sp->stream.open = _iostream_open;
......
...@@ -34,19 +34,32 @@ ...@@ -34,19 +34,32 @@
34 34
35 size_t mu_stream_default_buffer_size = MU_STREAM_DEFBUFSIZ; 35 size_t mu_stream_default_buffer_size = MU_STREAM_DEFBUFSIZ;
36 36
37 #define _stream_event(stream, code, n, p) \
38 do \
39 { \
40 if ((stream)->event_cb && \
41 ((stream)->event_mask & _MU_STR_EVMASK(code))) \
42 (stream)->event_cb (stream, code, n, p); \
43 } \
44 while (0)
45
46 static int _stream_read_unbuffered (mu_stream_t stream, void *buf, size_t size,
47 int full_read, size_t *pnread);
48 static int _stream_write_unbuffered (mu_stream_t stream,
49 const void *buf, size_t size,
50 int full_write, size_t *pnwritten);
51
37 static void 52 static void
38 _stream_setflag (struct _mu_stream *stream, int flag) 53 _stream_setflag (struct _mu_stream *stream, int flag)
39 { 54 {
40 if (stream->event_cb && (stream->event_mask & flag)) 55 _stream_event (stream, _MU_STR_EVENT_SETFLAG, flag, NULL);
41 stream->event_cb (stream, _MU_STR_EVENT_SET, flag);
42 stream->flags |= flag; 56 stream->flags |= flag;
43 } 57 }
44 58
45 static void 59 static void
46 _stream_clrflag (struct _mu_stream *stream, int flag) 60 _stream_clrflag (struct _mu_stream *stream, int flag)
47 { 61 {
48 if (stream->event_cb && (stream->event_mask & flag)) 62 _stream_event (stream, _MU_STR_EVENT_CLRFLAG, flag, NULL);
49 stream->event_cb (stream, _MU_STR_EVENT_CLR, flag);
50 stream->flags &= ~flag; 63 stream->flags &= ~flag;
51 } 64 }
52 65
...@@ -80,13 +93,12 @@ _mu_stream_seteof (mu_stream_t str) ...@@ -80,13 +93,12 @@ _mu_stream_seteof (mu_stream_t str)
80 _stream_setflag (str, _MU_STR_EOF); 93 _stream_setflag (str, _MU_STR_EOF);
81 } 94 }
82 95
83 #define _stream_advance_buffer(s,n) ((s)->cur += n, (s)->level -= n)
84 #define _stream_buffer_offset(s) ((s)->cur - (s)->buffer)
85 #define _stream_orig_level(s) ((s)->level + _stream_buffer_offset (s))
86 #define _stream_buffer_freespace(s) \ 96 #define _stream_buffer_freespace(s) \
87 ((s)->bufsize - (s)->level - _stream_buffer_offset(s)) 97 ((s)->bufsize - (s)->level)
88 #define _stream_buffer_is_full(s) (_stream_buffer_freespace(s) == 0) 98 #define _stream_buffer_is_full(s) (_stream_buffer_freespace(s) == 0)
89 99
100 #define _stream_curp(s) ((s)->buffer + (s)->pos)
101
90 static int 102 static int
91 _stream_fill_buffer (struct _mu_stream *stream) 103 _stream_fill_buffer (struct _mu_stream *stream)
92 { 104 {
...@@ -94,24 +106,25 @@ _stream_fill_buffer (struct _mu_stream *stream) ...@@ -94,24 +106,25 @@ _stream_fill_buffer (struct _mu_stream *stream)
94 size_t rdn; 106 size_t rdn;
95 int rc = 0; 107 int rc = 0;
96 char c; 108 char c;
97 109
110 stream->offset += stream->level;
98 switch (stream->buftype) 111 switch (stream->buftype)
99 { 112 {
100 case mu_buffer_none: 113 case mu_buffer_none:
101 return 0; 114 return 0;
102 115
103 case mu_buffer_full: 116 case mu_buffer_full:
104 rc = mu_stream_read_unbuffered (stream, 117 rc = _stream_read_unbuffered (stream,
105 stream->buffer, stream->bufsize, 118 stream->buffer, stream->bufsize,
106 0, 119 0,
107 &stream->level); 120 &stream->level);
108 break; 121 break;
109 122
110 case mu_buffer_line: 123 case mu_buffer_line:
111 for (n = 0; 124 for (n = 0;
112 n < stream->bufsize 125 n < stream->bufsize
113 && (rc = mu_stream_read_unbuffered (stream, 126 && (rc = _stream_read_unbuffered (stream,
114 &c, 1, 0, &rdn)) == 0;) 127 &c, 1, 0, &rdn)) == 0;)
115 { 128 {
116 if (rdn == 0) 129 if (rdn == 0)
117 { 130 {
...@@ -125,7 +138,13 @@ _stream_fill_buffer (struct _mu_stream *stream) ...@@ -125,7 +138,13 @@ _stream_fill_buffer (struct _mu_stream *stream)
125 stream->level = n; 138 stream->level = n;
126 break; 139 break;
127 } 140 }
128 stream->cur = stream->buffer; 141 if (rc == 0)
142 {
143 stream->offset -= stream->level;
144 stream->pos = 0;
145 _stream_event (stream, _MU_STR_EVENT_FILLBUF,
146 stream->level, _stream_curp (stream));
147 }
129 return rc; 148 return rc;
130 } 149 }
131 150
...@@ -139,7 +158,7 @@ _stream_buffer_full_p (struct _mu_stream *stream) ...@@ -139,7 +158,7 @@ _stream_buffer_full_p (struct _mu_stream *stream)
139 158
140 case mu_buffer_line: 159 case mu_buffer_line:
141 return _stream_buffer_is_full (stream) 160 return _stream_buffer_is_full (stream)
142 || memchr (stream->cur, '\n', stream->level) != NULL; 161 || memchr (stream->buffer, '\n', stream->level) != NULL;
143 162
144 case mu_buffer_full: 163 case mu_buffer_full:
145 return _stream_buffer_is_full (stream); 164 return _stream_buffer_is_full (stream);
...@@ -151,8 +170,9 @@ static int ...@@ -151,8 +170,9 @@ static int
151 _stream_flush_buffer (struct _mu_stream *stream, int all) 170 _stream_flush_buffer (struct _mu_stream *stream, int all)
152 { 171 {
153 int rc; 172 int rc;
154 char *end; 173 char *start, *end;
155 174 size_t wrsize;
175
156 if (stream->flags & _MU_STR_DIRTY) 176 if (stream->flags & _MU_STR_DIRTY)
157 { 177 {
158 if ((stream->flags & MU_STREAM_SEEK) && stream->seek) 178 if ((stream->flags & MU_STREAM_SEEK) && stream->seek)
...@@ -169,53 +189,55 @@ _stream_flush_buffer (struct _mu_stream *stream, int all) ...@@ -169,53 +189,55 @@ _stream_flush_buffer (struct _mu_stream *stream, int all)
169 abort(); /* should not happen */ 189 abort(); /* should not happen */
170 190
171 case mu_buffer_full: 191 case mu_buffer_full:
172 if ((rc = mu_stream_write_unbuffered (stream, stream->cur, 192 if ((rc = _stream_write_unbuffered (stream, stream->buffer,
173 stream->level, 1, NULL))) 193 stream->level, 1, NULL)))
174 return rc; 194 return rc;
175 _stream_advance_buffer (stream, stream->level); 195 _stream_event (stream, _MU_STR_EVENT_FLUSHBUF,
196 stream->level, stream->buffer);
176 break; 197 break;
177 198
178 case mu_buffer_line: 199 case mu_buffer_line:
179 if (stream->level == 0) 200 if (stream->level == 0)
180 break; 201 break;
181 for (end = memchr (stream->cur, '\n', stream->level); 202 wrsize = stream->level;
203 for (start = stream->buffer, end = memchr (start, '\n', wrsize);
182 end; 204 end;
183 end = memchr (stream->cur, '\n', stream->level)) 205 end = memchr (start, '\n', wrsize))
184 { 206 {
185 size_t size = end - stream->cur + 1; 207 size_t size = end - start + 1;
186 rc = mu_stream_write_unbuffered (stream, 208 rc = _stream_write_unbuffered (stream, start, size, 1, NULL);
187 stream->cur,
188 size, 1, NULL);
189 if (rc) 209 if (rc)
190 return rc; 210 return rc;
191 _stream_advance_buffer (stream, size); 211 _stream_event (stream, _MU_STR_EVENT_FLUSHBUF,
212 size, start);
213 start += size;
214 wrsize -= size;
215 if (wrsize == 0)
216 break;
192 } 217 }
193 if ((all && stream->level) || _stream_buffer_is_full (stream)) 218 if ((all && wrsize) || wrsize == stream->level)
194 { 219 {
195 rc = mu_stream_write_unbuffered (stream, 220 rc = _stream_write_unbuffered (stream,
196 stream->cur, 221 stream->buffer,
197 stream->level, 222 wrsize,
198 1, NULL); 223 1, NULL);
199 if (rc) 224 if (rc)
200 return rc; 225 return rc;
201 _stream_advance_buffer (stream, stream->level); 226 _stream_event (stream, _MU_STR_EVENT_FLUSHBUF,
227 wrsize, stream->buffer);
228 wrsize = 0;
202 } 229 }
230 if (wrsize)
231 memmove (stream->buffer, start, wrsize);
232 else
233 _stream_clrflag (stream, _MU_STR_DIRTY);
234 stream->level = stream->pos = wrsize;
235 return 0;
203 } 236 }
204 } 237 }
205 else if (all) 238
206 _stream_advance_buffer (stream, stream->level); 239 _stream_clrflag (stream, _MU_STR_DIRTY);
207 240 stream->pos = stream->level = 0;
208 if (stream->level)
209 {
210 if (stream->cur > stream->buffer)
211 memmove (stream->buffer, stream->cur, stream->level);
212 }
213 else
214 {
215 _stream_clrflag (stream, _MU_STR_DIRTY);
216 stream->level = 0;
217 }
218 stream->cur = stream->buffer;
219 return 0; 241 return 0;
220 } 242 }
221 243
...@@ -336,7 +358,7 @@ mu_stream_seek (mu_stream_t stream, mu_off_t offset, int whence, ...@@ -336,7 +358,7 @@ mu_stream_seek (mu_stream_t stream, mu_off_t offset, int whence,
336 case MU_SEEK_CUR: 358 case MU_SEEK_CUR:
337 if (offset == 0) 359 if (offset == 0)
338 { 360 {
339 *pres = stream->offset + _stream_buffer_offset (stream); 361 *pres = stream->offset + stream->pos;
340 return 0; 362 return 0;
341 } 363 }
342 offset += stream->offset; 364 offset += stream->offset;
...@@ -346,7 +368,7 @@ mu_stream_seek (mu_stream_t stream, mu_off_t offset, int whence, ...@@ -346,7 +368,7 @@ mu_stream_seek (mu_stream_t stream, mu_off_t offset, int whence,
346 rc = mu_stream_size (stream, &size); 368 rc = mu_stream_size (stream, &size);
347 if (rc) 369 if (rc)
348 return mu_stream_seterr (stream, rc, 1); 370 return mu_stream_seterr (stream, rc, 1);
349 offset += size; 371 offset += size + stream->pos;
350 break; 372 break;
351 373
352 default: 374 default:
...@@ -354,8 +376,9 @@ mu_stream_seek (mu_stream_t stream, mu_off_t offset, int whence, ...@@ -354,8 +376,9 @@ mu_stream_seek (mu_stream_t stream, mu_off_t offset, int whence,
354 } 376 }
355 377
356 if ((stream->buftype == mu_buffer_none && offset != stream->offset) 378 if ((stream->buftype == mu_buffer_none && offset != stream->offset)
379 || stream->level == 0
357 || offset < stream->offset 380 || offset < stream->offset
358 || offset > stream->offset + _stream_buffer_offset (stream)) 381 || offset > stream->offset + stream->level)
359 { 382 {
360 if ((rc = _stream_flush_buffer (stream, 1))) 383 if ((rc = _stream_flush_buffer (stream, 1)))
361 return rc; 384 return rc;
...@@ -366,9 +389,11 @@ mu_stream_seek (mu_stream_t stream, mu_off_t offset, int whence, ...@@ -366,9 +389,11 @@ mu_stream_seek (mu_stream_t stream, mu_off_t offset, int whence,
366 return mu_stream_seterr (stream, rc, 1); 389 return mu_stream_seterr (stream, rc, 1);
367 _mu_stream_cleareof (stream); 390 _mu_stream_cleareof (stream);
368 } 391 }
369 392 else if (stream->buftype != mu_buffer_none)
393 stream->pos = offset - stream->offset;
394
370 if (pres) 395 if (pres)
371 *pres = stream->offset + _stream_buffer_offset (stream); 396 *pres = stream->offset + stream->pos;
372 return 0; 397 return 0;
373 } 398 }
374 399
...@@ -386,56 +411,56 @@ static int ...@@ -386,56 +411,56 @@ static int
386 _stream_skip_input_bytes (mu_stream_t stream, mu_off_t count, mu_off_t *pres) 411 _stream_skip_input_bytes (mu_stream_t stream, mu_off_t count, mu_off_t *pres)
387 { 412 {
388 mu_off_t pos; 413 mu_off_t pos;
389 int rc; 414 int rc = 0;
390 415
391 if (!(stream->flags & MU_STREAM_READ)) 416 if (!(stream->flags & MU_STREAM_READ))
392 return mu_stream_seterr (stream, EACCES, 1); 417 return mu_stream_seterr (stream, EACCES, 1);
393 418
394 if (stream->buftype == mu_buffer_none) 419 if (count)
395 {
396 for (pos = 0; pos < count; pos++)
397 {
398 char c;
399 size_t nrd;
400 rc = mu_stream_read (stream, &c, 1, &nrd);
401 if (nrd == 0)
402 rc = ESPIPE;
403 if (rc)
404 break;
405 }
406 }
407 else
408 { 420 {
409 for (pos = 0;;) 421 if (stream->buftype == mu_buffer_none)
410 { 422 {
411 if ((rc = _stream_flush_buffer (stream, 1))) 423 for (pos = 0; pos < count; pos++)
412 return rc;
413 if (stream->level == 0)
414 { 424 {
415 rc = _stream_fill_buffer (stream); 425 char c;
426 size_t nrd;
427 rc = mu_stream_read (stream, &c, 1, &nrd);
428 if (nrd == 0)
429 rc = ESPIPE;
416 if (rc) 430 if (rc)
417 break; 431 break;
418 if (stream->level == 0) 432 }
433 }
434 else
435 {
436 for (pos = 0;;)
437 {
438 if ((rc = _stream_flush_buffer (stream, 1)))
439 return rc;
440 if (stream->pos == stream->level)
441 {
442 rc = _stream_fill_buffer (stream);
443 if (rc)
444 break;
445 if (stream->level == 0)
446 {
447 rc = ESPIPE;
448 break;
449 }
450 }
451 if (pos <= count && count < pos + stream->level)
419 { 452 {
420 rc = ESPIPE; 453 stream->pos = count - pos;
454 rc = 0;
421 break; 455 break;
422 } 456 }
457 pos += stream->level;
423 } 458 }
424 if (pos <= count && count < pos + stream->level)
425 {
426 size_t delta = count - pos;
427 _stream_advance_buffer (stream, delta);
428 pos = count;
429 rc = 0;
430 break;
431 }
432 pos += stream->level;
433 } 459 }
434 } 460 }
435 461
436 stream->offset += pos;
437 if (pres) 462 if (pres)
438 *pres = stream->offset; 463 *pres = stream->offset + stream->pos;
439 return rc; 464 return rc;
440 } 465 }
441 466
...@@ -482,7 +507,7 @@ mu_stream_set_buffer (mu_stream_t stream, enum mu_buffer_type type, ...@@ -482,7 +507,7 @@ mu_stream_set_buffer (mu_stream_t stream, enum mu_buffer_type type,
482 return mu_stream_seterr (stream, ENOMEM, 1); 507 return mu_stream_seterr (stream, ENOMEM, 1);
483 } 508 }
484 stream->bufsize = size; 509 stream->bufsize = size;
485 stream->cur = stream->buffer; 510 stream->pos = 0;
486 stream->level = 0; 511 stream->level = 0;
487 512
488 return 0; 513 return 0;
...@@ -496,10 +521,9 @@ mu_stream_get_buffer (mu_stream_t stream, struct mu_buffer_query *qry) ...@@ -496,10 +521,9 @@ mu_stream_get_buffer (mu_stream_t stream, struct mu_buffer_query *qry)
496 return 0; 521 return 0;
497 } 522 }
498 523
499 int 524 static int
500 mu_stream_read_unbuffered (mu_stream_t stream, void *buf, size_t size, 525 _stream_read_unbuffered (mu_stream_t stream, void *buf, size_t size,
501 int full_read, 526 int full_read, size_t *pnread)
502 size_t *pnread)
503 { 527 {
504 int rc; 528 int rc;
505 size_t nread; 529 size_t nread;
...@@ -560,11 +584,10 @@ mu_stream_read_unbuffered (mu_stream_t stream, void *buf, size_t size, ...@@ -560,11 +584,10 @@ mu_stream_read_unbuffered (mu_stream_t stream, void *buf, size_t size,
560 return rc; 584 return rc;
561 } 585 }
562 586
563 int 587 static int
564 mu_stream_write_unbuffered (mu_stream_t stream, 588 _stream_write_unbuffered (mu_stream_t stream,
565 const void *buf, size_t size, 589 const void *buf, size_t size,
566 int full_write, 590 int full_write, size_t *pnwritten)
567 size_t *pnwritten)
568 { 591 {
569 int rc; 592 int rc;
570 size_t nwritten; 593 size_t nwritten;
...@@ -624,7 +647,7 @@ int ...@@ -624,7 +647,7 @@ int
624 mu_stream_read (mu_stream_t stream, void *buf, size_t size, size_t *pread) 647 mu_stream_read (mu_stream_t stream, void *buf, size_t size, size_t *pread)
625 { 648 {
626 if (stream->buftype == mu_buffer_none) 649 if (stream->buftype == mu_buffer_none)
627 return mu_stream_read_unbuffered (stream, buf, size, !pread, pread); 650 return _stream_read_unbuffered (stream, buf, size, !pread, pread);
628 else 651 else
629 { 652 {
630 char *bufp = buf; 653 char *bufp = buf;
...@@ -634,7 +657,7 @@ mu_stream_read (mu_stream_t stream, void *buf, size_t size, size_t *pread) ...@@ -634,7 +657,7 @@ mu_stream_read (mu_stream_t stream, void *buf, size_t size, size_t *pread)
634 size_t n; 657 size_t n;
635 int rc; 658 int rc;
636 659
637 if (stream->level == 0) 660 if (stream->pos == stream->level)
638 { 661 {
639 if ((rc = _stream_fill_buffer (stream))) 662 if ((rc = _stream_fill_buffer (stream)))
640 { 663 {
...@@ -647,10 +670,10 @@ mu_stream_read (mu_stream_t stream, void *buf, size_t size, size_t *pread) ...@@ -647,10 +670,10 @@ mu_stream_read (mu_stream_t stream, void *buf, size_t size, size_t *pread)
647 } 670 }
648 671
649 n = size; 672 n = size;
650 if (n > stream->level) 673 if (n > stream->level - stream->pos)
651 n = stream->level; 674 n = stream->level - stream->pos;
652 memcpy (bufp, stream->cur, n); 675 memcpy (bufp, _stream_curp (stream), n);
653 _stream_advance_buffer (stream, n); 676 stream->pos += n;
654 nbytes += n; 677 nbytes += n;
655 bufp += n; 678 bufp += n;
656 size -= n; 679 size -= n;
...@@ -676,21 +699,24 @@ _stream_scandelim (mu_stream_t stream, char *buf, size_t size, int delim, ...@@ -676,21 +699,24 @@ _stream_scandelim (mu_stream_t stream, char *buf, size_t size, int delim,
676 return MU_ERR_BUFSPACE; 699 return MU_ERR_BUFSPACE;
677 while (size) 700 while (size)
678 { 701 {
679 char *p; 702 char *p, *q;
680 size_t len; 703 size_t len;
681 704
682 if (stream->level == 0) 705 if (stream->pos == stream->level)
683 { 706 {
684 if ((rc = _stream_fill_buffer (stream)) || stream->level == 0) 707 if ((rc = _stream_fill_buffer (stream)) || stream->level == 0)
685 break; 708 break;
686 } 709 }
687 710
688 p = memchr (stream->cur, delim, stream->level); 711 q = _stream_curp (stream);
689 len = p ? p - stream->cur + 1 : stream->level; 712 len = stream->level - stream->pos;
713 p = memchr (q, delim, len);
714 if (p)
715 len = p - q + 1;
690 if (len > size) 716 if (len > size)
691 len = size; 717 len = size;
692 memcpy (buf, stream->cur, len); 718 memcpy (buf, _stream_curp (stream), len);
693 _stream_advance_buffer (stream, len); 719 stream->pos += len;
694 buf += len; 720 buf += len;
695 size -= len; 721 size -= len;
696 nread += len; 722 nread += len;
...@@ -735,19 +761,22 @@ mu_stream_readdelim (mu_stream_t stream, char *buf, size_t size, ...@@ -735,19 +761,22 @@ mu_stream_readdelim (mu_stream_t stream, char *buf, size_t size,
735 761
736 if (size == 0) 762 if (size == 0)
737 return EINVAL; 763 return EINVAL;
738 764
739 if (stream->readdelim) 765 if (stream->buftype == mu_buffer_none)
740 { 766 {
741 size_t nread; 767 if (stream->readdelim)
742 rc = stream->readdelim (stream, buf, size, delim, &nread); 768 {
743 if (pread) 769 size_t nread;
744 *pread = nread; 770 rc = stream->readdelim (stream, buf, size, delim, &nread);
745 stream->offset += nread; 771 if (pread)
772 *pread = nread;
773 stream->offset += nread;
774 }
775 else
776 rc = _stream_readdelim (stream, buf, size, delim, pread);
746 } 777 }
747 else if (stream->buftype != mu_buffer_none)
748 rc = _stream_scandelim (stream, buf, size, delim, pread);
749 else 778 else
750 rc = _stream_readdelim (stream, buf, size, delim, pread); 779 rc = _stream_scandelim (stream, buf, size, delim, pread);
751 return rc; 780 return rc;
752 } 781 }
753 782
...@@ -847,8 +876,8 @@ mu_stream_write (mu_stream_t stream, const void *buf, size_t size, ...@@ -847,8 +876,8 @@ mu_stream_write (mu_stream_t stream, const void *buf, size_t size,
847 int rc = 0; 876 int rc = 0;
848 877
849 if (stream->buftype == mu_buffer_none) 878 if (stream->buftype == mu_buffer_none)
850 rc = mu_stream_write_unbuffered (stream, buf, size, 879 rc = _stream_write_unbuffered (stream, buf, size,
851 !pnwritten, pnwritten); 880 !pnwritten, pnwritten);
852 else 881 else
853 { 882 {
854 size_t nbytes = 0; 883 size_t nbytes = 0;
...@@ -868,8 +897,11 @@ mu_stream_write (mu_stream_t stream, const void *buf, size_t size, ...@@ -868,8 +897,11 @@ mu_stream_write (mu_stream_t stream, const void *buf, size_t size,
868 n = _stream_buffer_freespace (stream); 897 n = _stream_buffer_freespace (stream);
869 if (n > size) 898 if (n > size)
870 n = size; 899 n = size;
871 memcpy (stream->cur + stream->level, bufp, n); 900 memcpy (_stream_curp (stream), bufp, n);
872 stream->level += n; 901 stream->pos += n;
902 if (stream->pos > stream->level)
903 stream->level = stream->pos;
904
873 nbytes += n; 905 nbytes += n;
874 bufp += n; 906 bufp += n;
875 size -= n; 907 size -= n;
......
...@@ -279,7 +279,8 @@ mu_streamref_create_abridged (mu_stream_t *pref, mu_stream_t str, ...@@ -279,7 +279,8 @@ mu_streamref_create_abridged (mu_stream_t *pref, mu_stream_t str,
279 mu_stream_ref (str); 279 mu_stream_ref (str);
280 280
281 sp->stream.read = _streamref_read; 281 sp->stream.read = _streamref_read;
282 sp->stream.readdelim = _streamref_readdelim; 282 if (str->readdelim)
283 sp->stream.readdelim = _streamref_readdelim;
283 sp->stream.write = _streamref_write; 284 sp->stream.write = _streamref_write;
284 sp->stream.flush = _streamref_flush; 285 sp->stream.flush = _streamref_flush;
285 sp->stream.open = _streamref_open; 286 sp->stream.open = _streamref_open;
......
...@@ -167,21 +167,28 @@ print_transcript (struct _mu_xscript_stream *str, int flag, ...@@ -167,21 +167,28 @@ print_transcript (struct _mu_xscript_stream *str, int flag,
167 } 167 }
168 } 168 }
169 169
170 static int 170 static void
171 _xscript_read (struct _mu_stream *str, char *buf, size_t bufsize, 171 _xscript_event_cb (mu_stream_t str, int ev, unsigned long size, void *ptr)
172 size_t *pnread)
173 { 172 {
174 struct _mu_xscript_stream *sp = (struct _mu_xscript_stream *)str; 173 struct _mu_xscript_stream *sp = (struct _mu_xscript_stream *)str;
175 size_t nbytes;
176 int rc = mu_stream_read (sp->transport, buf, bufsize, &nbytes);
177 174
178 if (rc == 0) 175 switch (ev)
179 { 176 {
180 print_transcript (sp, TRANS_READ, buf, nbytes); 177 case _MU_STR_EVENT_FILLBUF:
181 if (pnread) 178 print_transcript (sp, TRANS_READ, ptr, size);
182 *pnread = nbytes; 179 break;
180
181 case _MU_STR_EVENT_FLUSHBUF:
182 print_transcript (sp, TRANS_WRITE, ptr, size);
183 } 183 }
184 return rc; 184 }
185
186 static int
187 _xscript_read (struct _mu_stream *str, char *buf, size_t bufsize,
188 size_t *pnread)
189 {
190 struct _mu_xscript_stream *sp = (struct _mu_xscript_stream *)str;
191 return mu_stream_read (sp->transport, buf, bufsize, pnread);
185 } 192 }
186 193
187 static int 194 static int
...@@ -189,15 +196,7 @@ _xscript_readdelim (struct _mu_stream *str, char *buf, size_t bufsize, ...@@ -189,15 +196,7 @@ _xscript_readdelim (struct _mu_stream *str, char *buf, size_t bufsize,
189 int delim, size_t *pnread) 196 int delim, size_t *pnread)
190 { 197 {
191 struct _mu_xscript_stream *sp = (struct _mu_xscript_stream *)str; 198 struct _mu_xscript_stream *sp = (struct _mu_xscript_stream *)str;
192 size_t nread; 199 return mu_stream_readdelim (sp->transport, buf, bufsize, delim, pnread);
193 int rc = mu_stream_readdelim (sp->transport, buf, bufsize, delim, &nread);
194 if (rc == 0)
195 {
196 print_transcript (sp, TRANS_READ, buf, nread);
197 if (pnread)
198 *pnread = nread;
199 }
200 return rc;
201 } 200 }
202 201
203 static int 202 static int
...@@ -205,11 +204,7 @@ _xscript_write (struct _mu_stream *str, const char *buf, size_t bufsize, ...@@ -205,11 +204,7 @@ _xscript_write (struct _mu_stream *str, const char *buf, size_t bufsize,
205 size_t *pnwrite) 204 size_t *pnwrite)
206 { 205 {
207 struct _mu_xscript_stream *sp = (struct _mu_xscript_stream *)str; 206 struct _mu_xscript_stream *sp = (struct _mu_xscript_stream *)str;
208 int rc = mu_stream_write (sp->transport, buf, bufsize, pnwrite); 207 return mu_stream_write (sp->transport, buf, bufsize, pnwrite);
209
210 if (rc == 0)
211 print_transcript (sp, TRANS_WRITE, buf, pnwrite ? *pnwrite : bufsize);
212 return rc;
213 } 208 }
214 209
215 static int 210 static int
...@@ -318,7 +313,6 @@ _xscript_ctl (struct _mu_stream *str, int op, void *arg) ...@@ -318,7 +313,6 @@ _xscript_ctl (struct _mu_stream *str, int op, void *arg)
318 case MU_IOCTL_GET_TRANSPORT_BUFFER: 313 case MU_IOCTL_GET_TRANSPORT_BUFFER:
319 case MU_IOCTL_SET_TRANSPORT_BUFFER: 314 case MU_IOCTL_SET_TRANSPORT_BUFFER:
320 { 315 {
321 struct mu_transport_buffer_query *qp = arg;
322 if (!sp->transport) 316 if (!sp->transport)
323 return EINVAL; 317 return EINVAL;
324 return mu_stream_ioctl (sp->transport, op, arg); 318 return mu_stream_ioctl (sp->transport, op, arg);
...@@ -391,7 +385,8 @@ mu_xscript_stream_create(mu_stream_t *pref, mu_stream_t transport, ...@@ -391,7 +385,8 @@ mu_xscript_stream_create(mu_stream_t *pref, mu_stream_t transport,
391 return ENOMEM; 385 return ENOMEM;
392 386
393 sp->stream.read = _xscript_read; 387 sp->stream.read = _xscript_read;
394 sp->stream.readdelim = _xscript_readdelim; 388 if (transport->readdelim)
389 sp->stream.readdelim = _xscript_readdelim;
395 sp->stream.write = _xscript_write; 390 sp->stream.write = _xscript_write;
396 sp->stream.flush = _xscript_flush; 391 sp->stream.flush = _xscript_flush;
397 sp->stream.open = _xscript_open; 392 sp->stream.open = _xscript_open;
...@@ -404,7 +399,9 @@ mu_xscript_stream_create(mu_stream_t *pref, mu_stream_t transport, ...@@ -404,7 +399,9 @@ mu_xscript_stream_create(mu_stream_t *pref, mu_stream_t transport,
404 sp->stream.truncate = _xscript_truncate; 399 sp->stream.truncate = _xscript_truncate;
405 sp->stream.shutdown = _xscript_shutdown; 400 sp->stream.shutdown = _xscript_shutdown;
406 sp->stream.error_string = _xscript_error_string; 401 sp->stream.error_string = _xscript_error_string;
407 402 sp->stream.event_cb = _xscript_event_cb;
403 sp->stream.event_mask = _MU_STR_EVMASK(_MU_STR_EVENT_FILLBUF) |
404 _MU_STR_EVMASK(_MU_STR_EVENT_FLUSHBUF);
408 if (!(flags & MU_STREAM_AUTOCLOSE)) 405 if (!(flags & MU_STREAM_AUTOCLOSE))
409 { 406 {
410 mu_stream_ref (transport); 407 mu_stream_ref (transport);
......