/pliant/language/stream/stream.pli
1	# Copyright  Hubert Tonneau  hubert.tonneau@pliant.cx
2	#
3	# This program is free software; you can redistribute it and/or
4	# modify it under the terms of the GNU General Public License version 2
5	# as published by the Free Software Foundation.
6	#
7	# This program is distributed in the hope that it will be useful,
8	# but WITHOUT ANY WARRANTY; without even the implied warranty of
9	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
10	# GNU General Public License for more details.
11	#
12	# You should have received a copy of the GNU General Public License
13	# version 2 along with this program; if not, write to the Free Software
14	# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
15
16	scope "/pliant/language/stream/" "/pliant/language/" "/pliant/protocol/dns/"
17	module "ring.pli"
18	module "/pliant/language/type/set/list.pli"
19
20	constant has_unreadline true
21	constant has_rewind true
22
23
24
25	# Stream
26
27
28	type StreamDriver
29	void
30
31	if has_rewind
32
33	type StreamBuffer
34	field Address start stop
35
36	type StreamMark
37	field Pointer:StreamBuffer buffer
38	field Int offset
39
40
41	type Stream
42	field Address read_cur read_stop
43	field Address write_cur write_stop
44
45	field Address read_buf
46	field Int read_buf_size
47	field Address write_buf
48	field Int write_buf_size
49
50	field Int flags
51	field Link:StreamDriver driver
52	field Str name
53	field ListNode_ list
54	field Int handle
55
56	field Int line_number
57	field Int line_limit
58	if has_unreadline
59	field Str next_line
60
61	if has_rewind
62	field List:StreamBuffer rewind_buffers
63	field Pointer:StreamBuffer rewind_current
64	field Address rewind_stop # backup of read_stop
65	field Array:StreamMark rewind_stack ; field Int rewind_count
66	field Int rewind_size rewind_limit
67
68	(addressof:Stream map Type) flags := Stream:flags .or. type_flag_do_not_copy
69
70
71
72
73
74	method fs open name options flags stream support -> status
75	oarg_rw FileSystem fs ; arg Str name options ; arg Int flags ; arg_rw Stream stream support ; arg ExtendedStatus status
76	generic
77	status := failure
78
79	method drv read buf mini maxi -> red
80	oarg_rw StreamDriver drv ; arg Address buf ; arg Int mini maxi red
81	generic
82	red := 0
83
84	method drv write buf mini maxi -> written
85	oarg_rw StreamDriver drv ; arg Address buf ; arg Int mini maxi written
86	generic
87	written := 0
88
89	method drv flush level -> status
90	oarg_rw StreamDriver drv ; arg Int level ; arg Status status
91	generic
92	status := success
93
94	method drv close -> status
95	oarg_rw StreamDriver drv ; arg ExtendedStatus status
96	generic
97	status := success
98
99	method drv query command stream answer -> status
100	oarg_rw StreamDriver drv ; arg Str command ; arg_rw Stream stream ; arg_w Str answer ; arg ExtendedStatus status
101	generic
102	answer := ""
103	status := failure
104
105	method drv configure command stream -> status
106	oarg_rw StreamDriver drv ; arg Str command ; arg_rw Stream stream ; arg ExtendedStatus status
107	generic
108	status := failure
109
110
111	export '. open'
112	export StreamDriver '. read' '. write' '. flush'
113	export Stream '. line_number' '. line_limit'
114
115
116
117	# other constants and global variables
118
119
120	constant crashed                01000000h
121	constant unflushed              02000000h
122	if has_unreadline
123	constant next_line_available  04000000h
124	if has_rewind
125	constant rewind_is_active    08000000h
126
127
128
129	# stream operations
130
131
132	method s is_open -> o
133	arg Stream s ; arg CBool o
134	o := (s:flags .and. in+out)<>0
135	check (shunt o (addressof s:driver)<>null (addressof s:driver)=null)
136
137
138	method s is_crashed -> c
139	arg Stream s ; arg CBool c
140	c := (s:flags .and. crashed)<>0
141
142	method s recover
143	arg_rw Stream s
144	s:flags := s:flags .and. .not. (cast crashed Int)
145
146
147	method s error msg
148	arg_rw Stream s ; arg Str msg
149	s flags := s:flags .or. crashed
150	s read_cur := s read_stop
151	s write_cur := s write_stop
152	if (s:flags .and. safe)=0
153	error error_id_io msg+" ("+s:name+")"
154
155
156	if has_rewind
157
158	method s rewind_read_required -> required
159	arg_rw Stream s ; arg CBool required
160	if (s:flags .and. rewind_is_active)=0
161	return true
162	var Pointer:StreamBuffer b1 :> s rewind_current
163	if exists:b1
164	var Pointer:StreamBuffer b2 :> s:rewind_buffers next s:rewind_current
165	s rewind_current :> b2
166	if exists:b2
167	s read_cur := b2 start
168	s read_stop := b2 stop
169	else
170	s read_cur := s read_buf
171	s read_stop := s rewind_stop
172	if s:rewind_count=0
173	memory_free b1:start
174	s rewind_size -= (cast b1:stop Int) .-. (cast b1:start Int)
175	s:rewind_buffers remove b1
176	if not exists:b2
177	s flags := s:flags .and. .not. (cast rewind_is_active Int)
178	required := s:read_cur=s:read_stop
179	else
180	var Int size := (cast s:read_stop Int) .-. (cast s:read_buf Int)
181	var Int skip := size
182	for (var Int i) 0 s:rewind_count-1
183	var Pointer:StreamMark m :> s:rewind_stack i
184	skip := shunt (exists m:buffer) 0 (min skip m:offset)
185	size -= skip
186	while s:rewind_size+size>s:rewind_limit and s:rewind_size>0
187	var Pointer:StreamBuffer b :> s:rewind_buffers first
188	for (var Int i) 0 s:rewind_count-1
189	var Pointer:StreamMark m :> s:rewind_stack i
190	if (addressof m:buffer)=addressof:b
191	m offset := undefined
192	memory_free b:start
193	s rewind_size -= (cast b:stop Int) .-. (cast b:start Int)
194	s:rewind_buffers remove b
195	if size>0
196	s rewind_buffers += var StreamBuffer empty_buffer
197	var Pointer:StreamBuffer b :> s:rewind_buffers last
198	b start := memory_allocate size addressof:s
199	memory_copy (s:read_buf translate Byte skip) b:start size
200	b stop := b:start translate Byte size
201	s rewind_size += size
202	for (var Int i) 0 s:rewind_count-1
203	var Pointer:StreamMark m :> s:rewind_stack i
204	if not (exists m:buffer)
205	m buffer :> b
206	m offset -= skip
207	else
208	for (var Int i) 0 s:rewind_count-1
209	var Pointer:StreamMark m :> s:rewind_stack i
210	if not (exists m:buffer)
211	m offset := 0
212	required := true
213
214	method s rewind_open
215	arg_rw Stream s
216	s flags := s:flags .or. rewind_is_active
217	var Int count := s rewind_count
218	s rewind_count := count+1
219	if count>=s:rewind_stack:size
220	s:rewind_stack size := count+1
221	var Pointer:StreamMark m :> s:rewind_stack count
222	m buffer :> s rewind_current
223	var Address start
224	if (exists s:rewind_current)
225	start := s:rewind_current start
226	else
227	start := s read_buf
228	m offset := (cast s:read_cur Int) .-. (cast start Int)
229
230	method s rewind
231	arg_rw Stream s
232	check s:rewind_count>0
233	var Pointer:StreamMark m :> s:rewind_stack s:rewind_count-1
234	if m:offset=undefined
235	s error "rewind capacity overflow"
236	return
237	var Pointer:StreamBuffer b :> m buffer
238	if not (exists s:rewind_current)
239	s rewind_stop := s read_stop
240	s rewind_current :> b
241	if exists:b
242	s read_cur := b:start translate Byte m:offset
243	s read_stop := b stop
244	else
245	s read_cur := s:read_buf translate Byte m:offset
246
247	method s rewind_close
248	arg_rw Stream s
249	var Int count := s rewind_count
250	check count>0
251	count -= 1
252	s rewind_count := count
253	if count=0
254	while { var Pointer:StreamBuffer b :> s:rewind_buffers first ; exists:b and addressof:b<>(addressof s:rewind_current) }
255	memory_free b:start
256	s rewind_size -= (cast b:stop Int) .-. (cast b:start Int)
257	s:rewind_buffers remove b
258	if not exists:b
259	s flags := s:flags .and. .not. (cast rewind_is_active Int)
260
261
262	method s reset
263	arg_rw Stream s
264	s read_cur := null ; s read_stop := null
265	s write_cur := null ; s write_stop := null
266	memory_free s:read_buf ; s read_buf := null
267	memory_free s:write_buf ; s write_buf := null
268	s flags := s:flags .and. crashed
269	s handle := undefined
270	s driver :> null map StreamDriver
271	s line_number := 0
272	s line_limit := 2^16
273	if has_rewind
274	each b s:rewind_buffers
275	memory_free b:start
276	s rewind_buffers := var List:StreamBuffer empty_buffers_list
277	s rewind_current :> null map StreamBuffer
278	s:rewind_stack size := 0 ; s rewind_count := 0
279	s rewind_size := 0
280	s rewind_limit := 2^24
281
282
283	method s write_all_data address size
284	arg_rw Stream s ; arg Address address ; arg Int size
285	var Address adr := address
286	var Int remain := size
287	while remain>0
288	var Int written := s:driver write adr remain remain
289	check written>=0
290	if written<=0
291	s error "Failed to write to stream"
292	return
293	else
294	s flags := s:flags .or. unflushed
295	adr := adr translate Byte written
296	remain := remain-written
297
298	method s flush level
299	arg_rw Stream s ; arg Int level
300	if (s:flags .and. out)=0
301	s error "Attempted to flush an "+(shunt s:is_open "in" "unopened")+" stream"
302	if s:is_crashed
303	return
304	if s:write_cur<>s:write_buf
305	s write_all_data s:write_buf (cast s:write_cur Int).-.(cast s:write_buf Int)
306	s write_cur := s write_buf
307	if (s:driver flush level)=failure
308	s error "Failed to flush stream ("+string:level+")"
309	s flags := s:flags .andnot. unflushed
310	(the_function '. flush' Stream Int) extra_module :> the_module "/pliant/language/stream/flushmode.pli"
311
312
313	method s query command -> answer
314	arg_rw Stream s ; arg Str command answer
315	if s:is_open
316	if (s:driver query command s answer)=failure
317	answer := ""
318	s error "Failed to query the stream (the query was "+command+")"
319	else
320	s error "Attempted to query an unopened stream"
321	answer := ""
322
323	method s safe_query command -> answer
324	arg_rw Stream s ; arg Str command answer
325	if s:is_open
326	if (s:driver query command s answer)=failure
327	answer := ""
328	else
329	answer := ""
330
331	method s configure command -> status
332	arg_rw Stream s ; arg Str command ; arg ExtendedStatus status
333	if s:is_open
334	status := s:driver configure command s
335	if status=failure
336	s error "Failed to configure the stream (the command was "+command+")"
337	else
338	s error "Attempted to configure an unopened stream"
339	status := failure
340
341	method s safe_configure command -> status
342	arg_rw Stream s ; arg Str command ; arg ExtendedStatus status
343	if s:is_open
344	status := s:driver configure command s
345	else
346	status := failure "Not an open stream"
347
348
349	method s raw_read address size
350	arg_rw Stream s ; arg Address address ; arg Int size
351	check size>=0
352	if (s:read_cur translate Byte size)<=s:read_stop
353	memory_copy s:read_cur address size
354	s read_cur := s:read_cur translate Byte size
355	else
356	if (s:flags .and. in)=0
357	s error "Attempted to read from an "+(shunt s:is_open "out" "unopened")+" stream"
358	if (s:flags .and. noautopost)=0 and (s:write_cur<>s:write_buf or (s:flags .and. unflushed)<>0)
359	s flush async
360	var Address adr := address
361	var Int remain := size
362	while remain>0
363	if s:is_crashed
364	memory_clear adr remain
365	return
366	eif s:read_cur<>s:read_stop
367	var Int step := min (cast s:read_stop Int).-.(cast s:read_cur Int) remain
368	memory_copy s:read_cur adr step
369	s read_cur := s:read_cur translate Byte step ; adr := adr translate Byte step ; remain := remain-step
370	eif remain<s:read_buf_size\2 or has_rewind and (s:flags .and. rewind_is_active)<>0
371	if not has_rewind or s:rewind_read_required
372	var Int red := s:driver read s:read_buf remain s:read_buf_size
373	check red>=0
374	if red<=0
375	s error "Failed to read from stream"
376	s read_cur := s read_buf ; s read_stop := s:read_buf translate Byte red
377	else
378	var Int red := s:driver read adr remain remain
379	check red>=0
380	if red<=0
381	s error "Failed to read from stream"
382	adr := adr translate Byte red ; remain := remain-red
383
384
385	method s raw_write address size
386	arg_rw Stream s ; arg Address address ; arg Int size
387	check size>=0
388	if (s:write_cur translate Byte size)<=s:write_stop
389	memory_copy address s:write_cur size
390	s write_cur := s:write_cur translate Byte size
391	else
392	if (s:flags .and. out)=0
393	s error "Attempted to write to an "+(shunt s:is_open "in" "unopened")+" stream"
394	if s:is_crashed
395	return
396	var Int more := min (cast s:write_stop Int).-.(cast s:write_cur Int) size
397	check more>=0
398	memory_copy address s:write_cur more
399	s write_cur := s:write_cur translate Byte more
400	s write_all_data s:write_buf (cast s:write_cur Int).-.(cast s:write_buf Int)
401	if (s:flags .and. crashed)<>0
402	return
403	s write_cur := s write_buf
404	var Address adr := address translate Byte more
405	var Int remain := size-more
406	check remain>=0
407	while remain>s:write_buf_size\2
408	var Int written := s:driver write adr remain-s:write_buf_size\2 remain
409	check written>=0
410	if written<=0
411	s error "Failed to write to stream"
412	return
413	else
414	s flags := s:flags .or. unflushed
415	adr := adr translate Byte written ; remain := remain-written
416	check remain>=0
417	memory_copy adr s:write_cur remain
418	s write_cur := s:write_cur translate Byte remain
419
420
421	method s close -> status
422	arg_rw Stream s ; arg ExtendedStatus status
423	if s:is_open
424	if (s:flags .and. out)<>0
425	s flush end
426	status := s:driver close
427	if status=failure
428	s error "Failed to close stream"
429	s reset
430	else
431	status := failure
432
433
434	method s open name options flags fs support -> status
435	arg_rw Stream s ; arg Str name options ; arg Int flags ; oarg_rw FileSystem fs ; arg_rw Stream support ; arg ExtendedStatus status
436	s close
437	s name := name
438	s flags := flags .or. (shunt (flags .and. append)=append out 0)
439	var Int cs := shunt (flags .and. nocache)<>0 0 (flags .and. linecache)<>0 2^8 (flags .and. bigcache)<>0 2^16 2^12
440	s read_buf_size := cs
441	s write_buf_size := cs
442	status := fs open name options s:flags s support
443	if status=failure and (s:flags .and. out+mkdir)=out+mkdir
444	if not (name:len>0 and name:0="[dq]" and (name parse (var Str base) any:(var Str opt)))
445	base := name ; opt := ""
446	var Int i := base:len
447	while { i := (base 0 i) search_last "/" 0 ; i<>0 and (fs configure string:(base 0 i+1) opt+(shunt opt:len<>0 and options:len<>0 " " "")+options "mkdir")=failure }
448	void
449	while { i := i+1 ; i := ((base i base:len) search "/" -i)+i ; i<>0 }
450	fs configure string:(base 0 i+1) opt+(shunt opt:len<>0 and options:len<>0 " " "")+options "mkdir"
451	status := fs open name options s:flags s support
452	if status=failure
453	i := 0
454	while { i := i+1 ; i := ((base i base:len) search "/" -i)+i ; i<>0 }
455	fs configure (base 0 i+1) opt+(shunt opt:len<>0 and options:len<>0 " " "")+options "mkdir"
456	status := fs open name options s:flags s support
457	if status=success
458	check (addressof s:driver)<>null
459	if (s:flags .and. in)<>0
460	s read_buf := memory_allocate s:read_buf_size addressof:s
461	s read_cur := s read_buf
462	s read_stop := s read_buf
463	else
464	s read_buf_size := 0
465	if (s:flags .and. out)<>0
466	s write_buf := memory_allocate s:write_buf_size addressof:s
467	s write_cur := s write_buf
468	s write_stop := s:write_buf translate Byte s:write_buf_size
469	else
470	s write_buf_size := 0
471	else
472	check (addressof s:driver)=null
473	s flags := crashed+(flags .and. safe)
474	(the_function '. open' Stream Str Str Int FileSystem Stream -> ExtendedStatus) extra_module :> the_module "/pliant/language/stream/openmode.pli"
475
476	method s open name flags -> status
477	arg_rw Stream s ; arg Str name ; arg Int flags ; arg ExtendedStatus status
478	status := s open name "" flags pliant_default_file_system (null map Stream)
479	(the_function '. open' Stream Str Int -> ExtendedStatus) extra_module :> the_module "/pliant/language/stream/openmode.pli"
480
481	method s open name options flags-> status
482	arg_rw Stream s ; arg Str name options ; arg Int flags ; arg ExtendedStatus status
483	status := s open name options flags pliant_default_file_system (null map Stream)
484	(the_function '. open' Stream Str Str Int -> ExtendedStatus) extra_module :> the_module "/pliant/language/stream/openmode.pli"
485
486
487	method s atend -> ae
488	arg_rw Stream s ; arg CBool ae
489	if s:read_cur<>s:read_stop
490	return false
491	if (s:flags .and. in)=0
492	s error "Attempted to test read end on an "+(shunt s:is_open "out" "unopened")+" stream"
493	if (s:flags .and. crashed)<>0
494	return true
495	eif has_unreadline and (s:flags .and. next_line_available)<>0
496	return false
497	else
498	if (s:flags .and. noautopost)=0 and (s:write_cur<>s:write_buf or (s:flags .and. unflushed)<>0)
499	s flush async
500	if (s:flags .and. crashed)<>0
501	return true
502	if not has_rewind or s:rewind_read_required
503	var Int red := s:driver read s:read_buf 1 s:read_buf_size
504	check red>=0
505	s read_cur := s read_buf
506	s read_stop := s:read_buf translate Byte red
507	return red<=0
508	else
509	return false
510
511
512	method s read_available address size
513	arg_rw Stream s ; arg_w Address address ; arg_w Int size
514	if not s:atend
515	address := s read_cur ; size := (cast s:read_stop Int).-.(cast s:read_cur Int)
516	s read_cur := s read_stop
517	else
518	address := null ; size := 0
519
520	method s read_available address size maxi
521	arg_rw Stream s ; arg_w Address address ; arg_w Int size ; arg Int maxi
522	if not s:atend
523	address := s read_cur ; size := min (cast s:read_stop Int).-.(cast s:read_cur Int) maxi
524	s read_cur := address translate Byte size
525	else
526	address := null ; size := 0
527
528
529	method s writechars chars
530	arg_rw Stream s ; arg Str chars
531	s raw_write chars:characters chars:len
532
533
534	method s readline -> l
535	arg_rw Stream s ; arg Str l
536	if has_unreadline and (s:flags .and. next_line_available)<>0
537	l := s next_line
538	s next_line := ""
539	s flags := s:flags-next_line_available
540	return
541	l := ""
542	if s:atend
543	return
544	s line_number := s:line_number+1
545	while true
546	if s:atend
547	return
548	var Int mode := s:flags .and. cr+lf
549	var Address eol := memory_search s:read_cur (cast s:read_stop Int).-.(cast s:read_cur Int) (shunt mode<>cr "[lf]" "[cr]"):characters Char:size
550	if  mode=0
551	var Address eol_cr := memory_search s:read_cur (cast s:read_stop Int).-.(cast s:read_cur Int) "[cr]":characters Char:size
552	var Address eol_lf := eol
553	eol :=  shunt eol_cr<>null and (eol_lf=null or (cast eol_cr Int).-.(cast s:read_cur Int)<(cast eol_lf Int).-.(cast s:read_cur Int)) eol_cr eol_lf
554	var Int extra := (cast (shunt eol<>null eol s:read_stop) Int) .-. (cast s:read_cur Int)
555	check extra>=0
556	if l:len+extra>s:line_limit
557	s error "too long line"
558	l := ""
559	return
560	l resize l:len+extra
561	memory_copy s:read_cur (l:characters translate Char l:len-extra) extra
562	if eol<>null
563	s read_cur := eol translate Char 1
564	if mode=cr+lf and l:len>0 and (l l:len-1)="[cr]"
565	l resize l:len-1
566	if mode=0
567	if eol=eol_cr and not s:atend and (s:read_cur map Char)="[lf]"
568	s read_cur := s:read_cur translate Char 1
569	mode := cr+lf
570	eif eol=eol_lf
571	mode := lf
572	else
573	check eol=eol_cr
574	mode := cr
575	if (s:flags .and. anyeol)=0
576	s:flags := s:flags .or. mode
577	return
578	else
579	s read_cur := s read_stop
580
581
582	if has_unreadline
583	method s unreadline l
584	arg_rw Stream s ; arg Str l
585	s next_line := l
586	s flags := s:flags .or. next_line_available
587
588
589	method s eol
590	arg_rw Stream s
591	var Int mode := s:flags .and. cr+lf
592	var Str eol := shunt mode=0 or mode=lf "[lf]" mode=cr "[cr]" "[cr][lf]"
593	s raw_write eol:characters eol:len
594	if (s:flags .and. linecache)<>0
595	s flush async
596
597
598	method s writeline l
599	arg_rw Stream s ; arg Str l
600	s writechars l
601	s eol
602
603
604	function raw_copy src dest mini maxi -> copied
605	arg_rw Stream src dest ; arg Int mini maxi copied
606	# should use the 2.2 sendfile function when the Linux kernel is >=2.2 and both streams have a handle
607	copied := 0
608	while copied<mini
609	src read_available (var Address adr) (var Int size) maxi-copied
610	if size=0
611	return
612	dest raw_write adr size
613	copied += size
614
615
616	function build s
617	arg_w Stream s
618	s read_buf := null
619	s write_buf := null
620	s flags := 0
621	s reset
622	s:list next :> s list
623
624	function destroy s
625	arg_w Stream s
626	s close
627
628	export '. open' '. close' '. raw_read' '. raw_write' '. flush' '. query' '. configure' '. safe_query' '. safe_configure'
629	export '. writechars' '. readline' '. eol' '. writeline'
630	if has_unreadline
631	export '. unreadline'
632	export '. atend' '. read_available' '. error'
633	export raw_copy
634
635
636
637
638
639	function 'cast Status' s -> stat
640	arg Stream s ; arg Status stat
641	explicit
642	stat := shunt s:is_open and not s:is_crashed success failure
643
644	export 'cast Status'
645
646
647	alias '. stream_handle' '. handle'
648	alias '. stream_flags' '. flags'
649	alias '. stream_driver' '. driver'
650	alias '. stream_read_buf' '. read_buf'
651	alias '. stream_read_buf_size' '. read_buf_size'
652	alias '. stream_read_cur' '. read_cur'
653	alias '. stream_read_stop' '. read_stop'
654	alias '. stream_write_buf' '. write_buf'
655	alias '. stream_write_buf_size' '. write_buf_size'
656	alias '. stream_write_cur' '. write_cur'
657	alias '. stream_write_stop' '. write_stop'
658	export '. stream_handle' '. stream_flags' '. stream_driver'
659	export '. stream_read_buf' '. stream_read_buf_size' '. stream_read_cur' '. stream_read_stop'
660	export '. stream_write_buf' '. stream_write_buf_size' '. stream_write_cur' '. stream_write_stop'
661	if has_rewind
662	export '. rewind_limit'
663	export '. rewind_open' '. rewind' '. rewind_close'
664
665	alias stream_flag_unflushed unflushed
666	alias stream_flag_crashed crashed
667	export stream_flag_unflushed stream_flag_crashed
668
669	export '. name' '. is_open' '. is_crashed' '. recover'