Megatest

Check-in [d8959da4df]
Login
Overview
Comment:Added none as transport option
Downloads: Tarball | ZIP archive | SQL archive
Timelines: family | ancestors | descendants | both | v2.0001-configurable-transport
Files: files | file ages | folders
SHA1: d8959da4df0a8e0788d130b334c3ae7735bee713
User & Date: matt on 2022-01-16 08:55:17
Other Links: branch diff | manifest | tags
Context
2022-01-16
12:24
no transport option basically working (db locks after a while). check-in: 63c4fcc524 user: matt tags: v2.0001-configurable-transport
08:55
Added none as transport option check-in: d8959da4df user: matt tags: v2.0001-configurable-transport
08:05
configurability working ok check-in: ac3127fbf2 user: matt tags: v2.0001-configurable-transport
Changes

Modified Makefile from [db842b7763] to [54802c9b97].

28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
# add dboard mtut and tcmt back later

# Configuration stuff
transport-flavor :
	@echo Creating transport-flavor with full as flavor. Options include: full, simple
	echo full > transport-flavor

ulex.scm dbmgrmod.scm : ulex.scm.template dbmgrmod.scm.template transport-flavor
	./configure

# module source files
MSRCFILES = autoload.scm dbi.scm ducttape-lib.scm pkts.scm stml2.scm	\
            cookie.scm mutils.scm mtargs.scm apimod.scm	ulex.scm	\
            configfmod.scm commonmod.scm dbmod.scm rmtmod.scm		\
            debugprint.scm mtver.scm csv-xml.scm servermod.scm		\







|







28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
# add dboard mtut and tcmt back later

# Configuration stuff
transport-flavor :
	@echo Creating transport-flavor with full as flavor. Options include: full, simple
	echo full > transport-flavor

ulex.scm dbmgrmod.scm : ulex.scm.template dbmgrmod.scm.template transport-flavor ulex-*/*scm
	./configure

# module source files
MSRCFILES = autoload.scm dbi.scm ducttape-lib.scm pkts.scm stml2.scm	\
            cookie.scm mutils.scm mtargs.scm apimod.scm	ulex.scm	\
            configfmod.scm commonmod.scm dbmod.scm rmtmod.scm		\
            debugprint.scm mtver.scm csv-xml.scm servermod.scm		\

Modified configure from [a623bd293e] to [18c7686f57].

12
13
14
15
16
17
18


19
20
21
22
23
24
25
#     Megatest is distributed in the hope that it will be useful,
#     but WITHOUT ANY WARRANTY; without even the implied warranty of
#     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#     GNU General Public License for more details.
# 
#     You should have received a copy of the GNU General Public License
#     along with Megatest.  If not, see <http://www.gnu.org/licenses/>.



# look at build.config (not a version controlled file and
# create ulex.scm and dbmgr.scm

if [[ -e transport-flavor ]];then
    FLAVOR=$(cat transport-flavor)
else







>
>







12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
#     Megatest is distributed in the hope that it will be useful,
#     but WITHOUT ANY WARRANTY; without even the implied warranty of
#     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#     GNU General Public License for more details.
# 
#     You should have received a copy of the GNU General Public License
#     along with Megatest.  If not, see <http://www.gnu.org/licenses/>.

# Flavors include: simple, full and none

# look at build.config (not a version controlled file and
# create ulex.scm and dbmgr.scm

if [[ -e transport-flavor ]];then
    FLAVOR=$(cat transport-flavor)
else

Modified dcommon.scm from [217250003b] to [a72cc297c4].

24
25
26
27
28
29
30

31
32
33
34
35
36
37
(declare (uses mtver))
(declare (uses debugprint))
(declare (uses commonmod))
(declare (uses configfmod))
(declare (uses rmtmod))
(declare (uses mtargs))
(declare (uses testsmod))


(module dcommon
	*

 (import scheme
	 chicken.base
	 chicken.condition







>







24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
(declare (uses mtver))
(declare (uses debugprint))
(declare (uses commonmod))
(declare (uses configfmod))
(declare (uses rmtmod))
(declare (uses mtargs))
(declare (uses testsmod))
(declare (uses dbmgrmod))

(module dcommon
	*

 (import scheme
	 chicken.base
	 chicken.condition
60
61
62
63
64
65
66

67
68
69
70
71
72
73
	 srfi-69
	 sparse-vectors
	 srfi-1
	 )

(import mtver
	dbmod

	commonmod
	debugprint
	configfmod
	rmtmod
	;; gutils
	(prefix mtargs args:)
	testsmod)







>







61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
	 srfi-69
	 sparse-vectors
	 srfi-1
	 )

(import mtver
	dbmod
	dbmgrmod
	commonmod
	debugprint
	configfmod
	rmtmod
	;; gutils
	(prefix mtargs args:)
	testsmod)

Modified rmtmod.scm from [23e97ebe5e] to [1b0587b637].

1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
			 #t))))
    (debug:print-info 4 *default-log-port* "starting exit process, finalizing databases.")
    (if (and no-hurry (debug:debug-mode 18))
	(rmt:print-db-stats))
    (let ((th1 (make-thread
		(lambda () ;; thread for cleaning up, give it five seconds
		  (let* ((start-time (current-seconds)))
		    (if *db-serv-info*
			(let* ((host (servdat-host *db-serv-info*))
			       (port 	       (servdat-port *db-serv-info*)))
			  (debug:print-info 0 *default-log-port* "Shutting down server/responder.")
			  ;;
			  ;; TODO - add flushing/waiting on the work queue
			  ;;
			  (rmt:server-shutdown host port)







|







1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
			 #t))))
    (debug:print-info 4 *default-log-port* "starting exit process, finalizing databases.")
    (if (and no-hurry (debug:debug-mode 18))
	(rmt:print-db-stats))
    (let ((th1 (make-thread
		(lambda () ;; thread for cleaning up, give it five seconds
		  (let* ((start-time (current-seconds)))
		    #;(if *db-serv-info*
			(let* ((host (servdat-host *db-serv-info*))
			       (port 	       (servdat-port *db-serv-info*)))
			  (debug:print-info 0 *default-log-port* "Shutting down server/responder.")
			  ;;
			  ;; TODO - add flushing/waiting on the work queue
			  ;;
			  (rmt:server-shutdown host port)

Modified tests/simplerun/tests/test1/testconfig from [db2c27ffef] to [b9db33e554].

22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

# Test requirements are specified here
[requirements]
# waiton setup
priority 0

# Iteration for your tests are controlled by the items section
[items]
# PARTOFDAY morning noon afternoon evening night

# test_meta is a section for storing additional data on your test
[test_meta]
author matt
owner  matt
description An example test







|







22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

# Test requirements are specified here
[requirements]
# waiton setup
priority 0

# Iteration for your tests are controlled by the items section
# [items]
# PARTOFDAY morning noon afternoon evening night

# test_meta is a section for storing additional data on your test
[test_meta]
author matt
owner  matt
description An example test

Added ulex-none/dbmgr.scm version [66aa956084].







































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
;;======================================================================
;; Copyright 2022, Matthew Welland.
;; 
;; This file is part of Megatest.
;; 
;;     Megatest is free software: you can redistribute it and/or modify
;;     it under the terms of the GNU General Public License as published by
;;     the Free Software Foundation, either version 3 of the License, or
;;     (at your option) any later version.
;; 
;;     Megatest is distributed in the hope that it will be useful,
;;     but WITHOUT ANY WARRANTY; without even the implied warranty of
;;     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
;;     GNU General Public License for more details.
;; 
;;     You should have received a copy of the GNU General Public License
;;     along with Megatest.  If not, see <http://www.gnu.org/licenses/>.

;;======================================================================

(declare (unit dbmgrmod))

(declare (uses ulex))
(declare (uses apimod))
(declare (uses pkts))
(declare (uses commonmod))
(declare (uses dbmod))
(declare (uses mtargs))
(declare (uses portloggermod))
(declare (uses debugprint))

(module dbmgrmod
    *

(import scheme
	chicken.base
	chicken.condition
	chicken.file
	chicken.format
	chicken.port
	chicken.process
	chicken.process-context
	chicken.process-context.posix
	chicken.sort
	chicken.string
	chicken.time
	
	(prefix sqlite3 sqlite3:)
	matchable
	md5
	message-digest
	regex
	s11n
	srfi-1
	srfi-18
	srfi-69
	system-information
	typed-records
	
	pkts
	ulex

	commonmod
	apimod
	dbmod
	debugprint
	(prefix mtargs args:)
	portloggermod
	)

;; ;; Configurations for server
;; ;; (tcp-buffer-size 2048)
;; ;; (max-connections 2048) 
;; 
;; ;; info about me as a listener and my connections to db servers
;; ;; stored (for now) in *db-serv-info*
;; ;;
;; (defstruct servdat
;;   (host #f)
;;   (port #f)
;;   (uuid #f)
;;   (dbfile #f)
;;   (uconn   #f) ;; this is the listener *FOR THIS PROCESS*
;;   (mode    #f)
;;   (status 'starting)
;;   (trynum 0) ;; count the number of ports we've tried
;;   (conns  (make-hash-table)) ;; apath/dbname => conndat
;;   ) 
;; 
;; (define *db-serv-info* (make-servdat))
;; 
;; (define (servdat->url sdat)
;;   (conc (servdat-host sdat)":"(servdat-port sdat)))
;; 
;; ;; db servers contact info
;; ;;
;; (defstruct conndat
;;   (apath    #f)
;;   (dbname   #f)
;;   (fullname #f)
;;   (hostport #f)
;;   (ipaddr   #f)
;;   (port     #f)
;;   (srvpkt   #f)
;;   (srvkey   #f)
;;   (lastmsg  0)
;;   (expires  0))
;; 
;; (define *srvpktspec*
;;   `((server (host    . h)
;; 	    (port    . p)
;; 	    (servkey . k)
;; 	    (pid     . i)
;; 	    (ipaddr  . a)
;; 	    (dbpath  . d))))
;; 
;; ;;======================================================================
;; ;;  S U P P O R T   F U N C T I O N S
;; ;;======================================================================
;; 
;; ;; set up the api proc, seems like there should be a better place for this?
;; ;;
;; ;; IS THIS NEEDED ANYMORE? TODO - REMOVE IF POSSIBLE
;; ;;
;; ;; (define api-proc (make-parameter conc))
;; ;; (api-proc api:execute-requests)
;; 
;; ;; do we have a connection to apath dbname and
;; ;; is it not expired? then return it
;; ;;
;; ;; else setup a connection
;; ;;
;; ;; if that fails, return '(#f "some reason") ;; NB// convert to raising an exception
;; ;;
;; (define (rmt:get-conn remdat apath dbname)
;;   (let* ((fullname (db:dbname->path apath dbname)))
;;     (hash-table-ref/default (servdat-conns remdat) fullname #f)))
;; 
;; (define (rmt:drop-conn remdat apath dbname)
;;   (let* ((fullname (db:dbname->path apath dbname)))
;;     (hash-table-delete! (servdat-conns remdat) fullname)))
;;   
;; (define (rmt:find-main-server uconn apath dbname)
;;   (let* ((pktsdir     (get-pkts-dir apath))
;; 	 (all-srvpkts (get-all-server-pkts pktsdir *srvpktspec*))
;; 	 (viable-srvs (get-viable-servers all-srvpkts dbname)))
;;     (get-the-server uconn apath viable-srvs)))
;; 
;; 
;; (define *connstart-mutex* (make-mutex))
;; (define *last-main-start* 0)
;; 
;; ;; looks for a connection to main, returns if have and not exired
;; ;; creates new otherwise
;; ;; 
;; ;; connections for other servers happens by requesting from main
;; ;;
;; ;; TODO: This is unnecessarily re-creating the record in the hash table
;; ;;
;; (define (rmt:open-main-connection remdat apath)
;;   (let* ((fullpath (db:dbname->path apath ".db/main.db"))
;; 	 (conns    (servdat-conns remdat))
;; 	 (conn     (rmt:get-conn remdat apath ".db/main.db")) ;; (hash-table-ref/default conns fullpath #f)) ;; TODO - create call for this
;; 	 (start-rmt:run (lambda ()
;; 			  (let* ((th1 (make-thread (lambda ()(rmt:run (get-host-name))) "non-db mode server")))
;; 			    (thread-start! th1)
;; 			    (thread-sleep! 1)
;; 			    (let loop ((count 0))
;; 			      (assert (< count 30) "FATAL: responder failed to initialize in rmt:open-main-connection")
;; 			      (if (or (not *db-serv-info*)
;; 				      (not (servdat-uconn *db-serv-info*)))
;; 				  (begin
;; 				    (thread-sleep! 1)
;; 				    (loop (+ count 1)))
;; 				  (begin
;; 				    (servdat-mode-set! *db-serv-info* 'non-db)
;; 				    (servdat-uconn *db-serv-info*)))))))
;; 	 (myconn    (servdat-uconn *db-serv-info*)))
;;     (cond
;;      ((not myconn)
;;       (start-rmt:run)
;;       (rmt:open-main-connection remdat apath))
;;      ((and conn                                             ;; conn is NOT a socket, just saying ...
;; 	   (< (current-seconds) (conndat-expires conn)))
;;       #t) ;; we are current and good to go - we'll deal elsewhere with a server that was killed or died 
;;      ((and conn
;; 	   (>= (current-seconds)(conndat-expires conn)))
;;       (debug:print-info 0 *default-log-port* "connection to "fullpath" server expired. Reconnecting.")
;;       (rmt:drop-conn remdat apath ".db/main.db") ;;
;;       (rmt:open-main-connection remdat apath))
;;      (else
;;       ;; Below we will find or create and connect to main
;;       (debug:print-info 0 *default-log-port* "rmt:open-main-connection - starting from scratch")
;;       (let* ((dbname         (db:run-id->dbname #f))
;; 	     (the-srv        (rmt:find-main-server myconn apath dbname))
;; 	     (start-main-srv (lambda () ;; call IF there is no the-srv found
;; 			       (mutex-lock! *connstart-mutex*)
;; 			       (if (> (- (current-seconds) *last-main-start*) 5) ;; at least four seconds since last attempt to start main server
;; 				   (begin
;; 				     (api:run-server-process apath dbname)
;; 				     (set! *last-main-start* (current-seconds))
;; 				     (thread-sleep! 1))
;; 				   (thread-sleep! 0.25))
;; 			       (mutex-unlock! *connstart-mutex*)
;; 			       (rmt:open-main-connection remdat apath) ;; TODO: Add limit to number of tries
;; 			       )))
;; 	(if (not the-srv) ;; have server, try connecting to it
;; 	    (start-main-srv)
;; 	    (let* ((srv-addr (server-address the-srv)) ;; need serv
;; 		   (ipaddr   (alist-ref 'ipaddr  the-srv))
;; 		   (port     (alist-ref 'port    the-srv))
;; 		   (srvkey   (alist-ref 'servkey the-srv))
;; 		   (fullpath (db:dbname->path apath dbname))
;; 		   
;; 		   (new-the-srv (make-conndat
;; 				 apath:   apath
;; 				 dbname:  dbname
;; 				 fullname: fullpath
;; 				 hostport: srv-addr
;; 				 ;; socket: (open-nn-connection srv-addr)  - TODO - open ulex connection?
;; 				 ipaddr: ipaddr
;; 				 port: port
;; 				 srvpkt: the-srv
;; 				 srvkey: srvkey ;; generated by rmt:get-signature on the server side
;; 				 lastmsg: (current-seconds)
;; 				 expires: (+ (current-seconds)
;; 					     (server:expiration-timeout)
;; 					     -2) ;; this needs to be gathered during the ping
;; 				 )))
;; 	      (hash-table-set! conns fullpath new-the-srv)))
;; 	#t)))))
;; 
;; ;; NB// sinfo is a servdat struct
;; ;;
;; (define (rmt:general-open-connection sinfo apath dbname #!key (num-tries 5))
;;   (assert (not (equal? dbname ".db/main.db")) "ERROR: general-open-connection should never be called with main as the db")
;;   (let* ((mdbname  ".db/main.db") ;; (db:run-id->dbname #f)) TODO: put this back to the lookup when stable
;; 	 (fullname (db:dbname->path apath dbname))
;; 	 (conns    (servdat-conns sinfo))
;; 	 (mconn    (rmt:get-conn sinfo apath ".db/main.db"))
;; 	 (dconn    (rmt:get-conn sinfo apath dbname)))
;;     #;(if (and mconn
;; 	     (not (debug:print-logger)))
;; 	(begin
;; 	  (debug:print-info 0 *default-log-port* "Turning on logging to main, look in logs dir for main log.")
;; 	  (debug:print-logger rmt:log-to-main)))
;;     (cond
;;      ((and mconn
;; 	   dconn
;; 	   (< (current-seconds)(conndat-expires dconn)))
;;       #t) ;; good to go
;;      ((not mconn) ;; no channel open to main? open it...
;;       (rmt:open-main-connection sinfo apath)
;;       (rmt:general-open-connection sinfo apath dbname num-tries: (- num-tries 1)))
;;      ((not dconn)                 ;; no channel open to dbname?     
;;       (let* ((res (rmt:send-receive-real sinfo apath mdbname 'get-server `(,apath ,dbname))))
;; 	(case res
;; 	  ((server-started)
;; 	   (if (> num-tries 0)
;; 	       (begin
;; 		 (thread-sleep! 2)
;; 		 (rmt:general-open-connection sinfo apath dbname num-tries: (- num-tries 1)))
;; 	       (begin
;; 		 (debug:print-error 0 *default-log-port* "Failed to start servers needed or open channel to "apath", "dbname)
;; 		 (exit 1))))
;; 	  (else
;; 	   (if (list? res) ;; server has been registered and the info was returned. pass it on.
;; 	       (begin ;;  ("192.168.0.9" 53817
;; 		      ;;  "5e34239f48e8973b3813221e54701a01" "24310"
;; 		      ;;  "192.168.0.9"
;; 		      ;;  "/home/matt/data/megatest/tests/simplerun"
;; 		 ;;  ".db/1.db")
;; 		 (match
;; 		  res
;; 		  ((host port servkey pid ipaddr apath dbname)
;; 		   (debug:print-info 0 *default-log-port* "got "res)
;; 		   (hash-table-set! conns
;; 				    fullname
;; 				    (make-conndat
;; 				     apath: apath
;; 				     dbname: dbname
;; 				     hostport: (conc host":"port)
;; 				     ;; socket: (open-nn-connection (conc host":"port)) ;; TODO - open ulex connection?
;; 				     ipaddr: ipaddr
;; 				     port: port
;; 				     srvkey: servkey
;; 				     lastmsg: (current-seconds)
;; 				     expires: (+ (current-seconds)
;; 						 (server:expiration-timeout)
;; 						 -2))))
;; 		  (else
;; 		   (debug:print-info 0 *default-log-port* "return data from starting server did not match host port servkey pid ipaddr apath dbname " res)))
;; 		 res)
;; 	       (begin
;; 		 (debug:print-info 0 *default-log-port* "Unexpected result: " res)
;; 		 res)))))))
;;     #t))
;; 
;; ;;======================================================================
;; 
;; ;; FOR DEBUGGING SET TO #t
;; ;; (define *localmode* #t)
;; (define *localmode* #f)
(define *dbstruct* (make-dbr:dbstruct))
 
;; Defaults to current area
;;
(define (rmt:send-receive cmd rid params #!key (attemptnum 1)(area-dat #f))
  (let* ((apath      *toppath*)
	 (dbname     (db:run-id->dbname rid)))
	(api:execute-requests *dbstruct* cmd params)))

;; ;; db is at apath/.db/dbname, rid is an intermediary solution and will be removed
;; ;; sometime in the future
;; ;;
;; (define (rmt:send-receive-real sinfo apath dbname cmd params)
;;   (assert (not (eq? 'primordial (thread-name (current-thread)))) "FATAL: Do not call rmt:send-receive-real in the primodial thread.")
;;   (let* ((cdat (rmt:get-conn sinfo apath dbname)))
;;     (assert cdat "FATAL: rmt:send-receive-real called without the needed channels opened")
;;     (let* ((uconn    (servdat-uconn sinfo)) ;; get the interface to ulex
;; 	   ;; then send-receive using the ulex layer to host-port stored in cdat
;; 	   (res      (send-receive uconn (conndat-hostport cdat) cmd params))
;; 	   #;(th1      (make-thread (lambda ()
;; 				    (set! res (send-receive uconn (conndat-hostport cdat) cmd params)))
;; 				  "send-receive thread")))
;;       ;; (thread-start! th1)
;;       ;; (thread-join! th1)   ;; gratuitious thread stuff is so that mailbox is not used in primordial thead
;;       ;; since we accessed the server we can bump the expires time up
;;       (conndat-expires-set! cdat (+ (current-seconds)
;; 				    (server:expiration-timeout)
;; 				    -2)) ;; two second margin for network time misalignments etc.
;;       res)))
;; 
;; db is at apath/.db/dbname, rid is an intermediary solution and will be removed
;; sometime in the future.
;;
;; Purpose - call the main.db server and request a server be started
;; for the given area path and dbname
;;

(define (rmt:print-db-stats)
  (let ((fmtstr "~40a~7-d~9-d~20,2-f")) ;; "~20,2-f"
    (debug:print 18 *default-log-port* "DB Stats, "(seconds->year-week/day-time (current-seconds))"\n=====================")
    (debug:print 18 *default-log-port* (format #f "~40a~8a~10a~10a" "Cmd" "Count" "TotTime" "Avg"))
    (for-each (lambda (cmd)
		(let ((cmd-dat (hash-table-ref *db-stats* cmd)))
		  (debug:print 18 *default-log-port* (format #f fmtstr cmd (vector-ref cmd-dat 0) (vector-ref cmd-dat 1) (/ (vector-ref cmd-dat 1)(vector-ref cmd-dat 0))))))
	      (sort (hash-table-keys *db-stats*)
		    (lambda (a b)
		      (> (vector-ref (hash-table-ref *db-stats* a) 0)
			 (vector-ref (hash-table-ref *db-stats* b) 0)))))))

(define (rmt:get-max-query-average run-id)
  (mutex-lock! *db-stats-mutex*)
  (let* ((runkey (conc "run-id=" run-id " "))
	 (cmds   (filter (lambda (x)
			   (substring-index runkey x))
			 (hash-table-keys *db-stats*)))
	 (res    (if (null? cmds)
		     (cons 'none 0)
		     (let loop ((cmd (car cmds))
				(tal (cdr cmds))
				(max-cmd (car cmds))
				(res 0))
		       (let* ((cmd-dat (hash-table-ref *db-stats* cmd))
			      (tot     (vector-ref cmd-dat 0))
			      (curravg (/ (vector-ref cmd-dat 1) (vector-ref cmd-dat 0))) ;; count is never zero by construction
			      (currmax (max res curravg))
			      (newmax-cmd (if (> curravg res) cmd max-cmd)))
			 (if (null? tal)
			     (if (> tot 10)
				 (cons newmax-cmd currmax)
				 (cons 'none 0))
			     (loop (car tal)(cdr tal) newmax-cmd currmax)))))))
    (mutex-unlock! *db-stats-mutex*)
    res))

;; ;; host and port are used to ensure we are remove proper records
;; (define (rmt:server-shutdown host port)
;;   (let ((dbfile   (servdat-dbfile *db-serv-info*)))
;;     (debug:print-info 0 *default-log-port* "dbfile is "dbfile)
;;     (if dbfile
;; 	(let* ((am-server  (args:get-arg "-server"))
;; 	       (dbfile     (args:get-arg "-db"))
;; 	       (apath      *toppath*)
;; 	       #;(sinfo     *remotedat*)) ;; foundation for future fix
;; 	  (if *dbstruct-db*
;; 	      (let* ((dbdat      (db:get-dbdat *dbstruct-db* apath dbfile))
;; 		     (db         (dbr:dbdat-db dbdat))
;; 		     (inmem      (dbr:dbdat-db dbdat))   ;; WRONG
;; 		     )
;; 		;; do a final sync here
;; 		(debug:print-info 0 *default-log-port* "Doing final sync for "apath" "dbfile" at "(current-seconds))
;; 		(db:sync-inmem->disk *dbstruct-db* apath dbfile force-sync: #t)
;; 		;; let's finalize here
;; 		(debug:print-info 0 *default-log-port* "Finalizing db and inmem")
;; 		(if (sqlite3:database? db)
;; 		    (sqlite3:finalize! db)
;; 		    (debug:print-info 0 *default-log-port* "in rmt:server-shutdown, db is not a database, not finalizing..."))
;; 		(if (sqlite3:database? inmem)
;; 		    (sqlite3:finalize! inmem)
;; 		    (debug:print-info 0 *default-log-port* "in rmt:server-shutdown, inmem is not a database, not finalizing..."))
;; 		(debug:print-info 0 *default-log-port* "Finalizing db and inmem complete"))
;; 	      (debug:print-info 0 *default-log-port* "Db was never opened, no cleanup to do."))
;; 	  (if (not am-server)
;; 	      (debug:print-info 0 *default-log-port* "I am not a server, should NOT get here!")
;; 	      (if (string-match ".*/main.db$" dbfile)
;; 		  (let ((pkt-file (conc (get-pkts-dir *toppath*)
;; 					"/" (servdat-uuid *db-serv-info*)
;; 					".pkt")))
;; 		    (debug:print-info 0 *default-log-port* "removing pkt "pkt-file)
;; 		    (delete-file* pkt-file)
;; 		    (debug:print-info 0 *default-log-port* "Releasing lock (if any) for "dbfile ", host "host", port "port)
;; 		    (db:with-lock-db
;; 		     (servdat-dbfile *db-serv-info*)
;; 		     (lambda (dbh dbfile)
;; 		       (db:release-lock dbh dbfile host port)))) ;; I'm not the server - should not have a lock to remove
;; 		  (let* ((sdat *db-serv-info*) ;; we have a run-id server
;; 			 (host (servdat-host sdat))
;; 			 (port (servdat-port sdat))
;; 			 (uuid (servdat-uuid sdat))
;; 			 (res  (rmt:deregister-server *db-serv-info* *toppath* host port uuid dbfile)))
;; 		    (debug:print-info 0 *default-log-port* "deregistered-server, res="res)
;; 		    (debug:print-info 0 *default-log-port* "deregistering server "host":"port" with uuid "uuid)
;; 		    )))))))
;; 
;; 
;; (define (common:run-sync?)
;;     ;; (and (common:on-homehost?)
;;   (args:get-arg "-server"))
;; 
;; (define *rmt:run-mutex* (make-mutex))
;; (define *rmt:run-flag* #f)
;; 
;; ;; Main entry point to start a server. was start-server
;; (define (rmt:run hostn)
;;   (mutex-lock! *rmt:run-mutex*)
;;   (if *rmt:run-flag*
;;       (begin
;; 	(debug:print-warn 0 *default-log-port* "rmt:run already running.")
;; 	(mutex-unlock! *rmt:run-mutex*))
;;       (begin
;; 	(set! *rmt:run-flag* #t)
;; 	(mutex-unlock! *rmt:run-mutex*)
;; 	;;  ;; Configurations for server
;; 	;;  (tcp-buffer-size 2048)
;; 	;;  (max-connections 2048) 
;; 	(debug:print 2 *default-log-port* "PID: "(current-process-id)". Attempting to start the server ...")
;; 	(if (and *db-serv-info*
;; 		 (servdat-uconn *db-serv-info*))
;; 	    (let* ((uconn (servdat-uconn *db-serv-info*)))
;; 	      (wait-and-close uconn))
;; 	    (let* ((port            (portlogger:open-run-close portlogger:find-port))
;; 		   (handler-proc    (lambda (rem-host-port qrykey cmd params) ;;
;; 				      (set! *db-last-access* (current-seconds))
;; 				      (assert (list? params) "FATAL: handler called with non-list params")
;; 				      (assert (args:get-arg "-server") "FATAL: handler called on non-server side. cmd="cmd", params="params)
;; 				      (debug:print 0 *default-log-port* "handler call: "cmd", params="params)
;; 				      (api:execute-requests *dbstruct-db* cmd params))))
;; 	      ;; (api:process-request *dbstuct-db* 
;; 	      (if (not *db-serv-info*)
;; 		  (set! *db-serv-info* (make-servdat host: hostn port: port)))
;; 	      (let* ((uconn (run-listener handler-proc port))
;; 		     (rport (udat-port uconn))) ;; the real port
;; 		(servdat-host-set! *db-serv-info* hostn)
;; 		(servdat-port-set! *db-serv-info* rport)
;; 		(servdat-uconn-set! *db-serv-info* uconn)
;; 		(wait-and-close uconn)
;; 		(db:print-current-query-stats)
;; 		)))
;; 	(let* ((host (servdat-host *db-serv-info*))
;; 	       (port (servdat-port *db-serv-info*))
;; 	       (mode (or (servdat-mode *db-serv-info*)
;; 			 "non-db")))
;; 	  ;; server exit stuff here
;; 	  ;; (rmt:server-shutdown host port) - always do in on-exit
;; 	  ;; (portlogger:open-run-close portlogger:set-port port "released") ;; moved to on-exit 
;; 	  (debug:print-info 0 *default-log-port* "Server "host":"port" mode "mode"shutdown complete. Exiting")
;; 	  ))))
;; 
;; ;;======================================================================
;; ;; S E R V E R   U T I L I T I E S 
;; ;;======================================================================
;; 
;; 
;; ;;======================================================================
;; ;; NEW SERVER METHOD
;; ;;======================================================================
;; 
;; ;; only use for main.db - need to re-write some of this :(
;; ;;
;; (define (get-lock-db sdat dbfile host port)
;;   (assert host "FATAL: get-lock-db called with host not set.")
;;   (assert port "FATAL: get-lock-db called with port not set.")
;;   (let* ((dbh (db:open-run-db dbfile db:initialize-db)) ;; open-run-db creates a standard db with schema used by all situations
;; 	 (res (db:get-iam-server-lock dbh dbfile host port))
;; 	 (uconn (servdat-uconn sdat)))
;;     ;; res => list then already locked, check server is responsive
;;     ;;     => #t then sucessfully got the lock
;;     ;;     => #f reserved for future use as to indicate something went wrong
;;     (match res
;;       ((owner_pid owner_host owner_port event_time)
;;        (if (server-ready? uconn (conc owner_host":"owner_port) "abc")
;; 	   #f      ;; locked by someone else
;; 	   (begin  ;; locked by someone dead and gone
;; 	     (debug:print 0 *default-log-port* "WARNING: stale lock - have to steal it. This may fail.")
;; 	     (db:steal-lock-db dbh dbfile port))))
;;       (#t  #t) ;; placeholder so that we don't touch res if it is #t
;;       (else (set! res #f)))
;;     (sqlite3:finalize! dbh)
;;     res))
;; 
;; 
;; (define (register-server pkts-dir pkt-spec host port servkey ipaddr dbpath)
;;   (let* ((pkt-dat `((host    . ,host)
;; 		    (port    . ,port)
;; 		    (servkey . ,servkey)
;; 		    (pid     . ,(current-process-id))
;; 		    (ipaddr  . ,ipaddr)
;; 		    (dbpath  . ,dbpath)))
;; 	 (uuid    (write-alist->pkt
;; 		   pkts-dir
;; 		   pkt-dat
;; 		   pktspec: pkt-spec
;; 		   ptype: 'server)))
;;     (debug:print 0 *default-log-port* "Server on "host":"port" registered in pkt "uuid)
;;     uuid))
;; 
;; (define (get-pkts-dir #!optional (apath #f))
;;   (let* ((effective-toppath (or *toppath* apath)))
;;     (assert effective-toppath
;; 	    "ERROR: get-pkts-dir called without *toppath* set. Exiting.")
;;     (let* ((pdir (conc effective-toppath "/.meta/srvpkts")))
;;       (if (file-exists? pdir)
;; 	  pdir
;; 	  (begin
;; 	    (handle-exceptions ;; this exception handler should NOT be needed but ...
;; 		exn
;; 		pdir
;; 	      (create-directory pdir #t))
;; 	    pdir)))))
;; 
;; ;; given a pkts dir read 
;; ;;
;; (define (get-all-server-pkts pktsdir-in pktspec)
;;   (let* ((pktsdir  (if (file-exists? pktsdir-in)
;; 		       pktsdir-in
;; 		       (begin
;; 			 (create-directory pktsdir-in #t)
;; 			 pktsdir-in)))
;; 	 (all-pkt-files (glob (conc pktsdir "/*.pkt"))))
;;     (map (lambda (pkt-file)
;; 	   (read-pkt->alist pkt-file pktspec: pktspec))
;; 	 all-pkt-files)))
;; 
;; (define (server-address srv-pkt)
;;   (conc (alist-ref 'host srv-pkt) ":"
;; 	(alist-ref 'port srv-pkt)))
	
(define (server-ready? uconn host-port key) ;; server-address is host:port
  #;(let* ((params `((cmd . ping)(key . ,key)))
	 (data `((cmd . ping)
		 (key . ,key)
		 (params . ,params))) ;; I don't get it.
	 (res  (send-receive uconn host-port 'ping data)))
    (if (eq? res 'ack) ;; yep, likely it is who we want on the other end
	res
  #f))
  #t)

;; ; from the pkts return servers associated with dbpath
;; ;; NOTE: Only one can be alive - have to check on each
;; ;;       in the list of pkts returned
;; ;;
;; (define (get-viable-servers serv-pkts dbpath)
;;   (let loop ((tail serv-pkts)
;; 	     (res  '()))
;;     (if (null? tail)
;; 	res ;; NOTE: sort by age so oldest is considered first
;; 	(let* ((spkt (car tail)))
;; 	  (loop (cdr tail)
;; 		(if (equal? dbpath (alist-ref 'dbpath spkt))
;; 		    (cons spkt res)
;; 		    res))))))
;; 
;; (define (remove-pkts-if-not-alive uconn serv-pkts)
;;   (filter (lambda (pkt)
;; 	    (let* ((host (alist-ref 'host pkt))
;; 		   (port (alist-ref 'port pkt))
;; 		   (host-port (conc host":"port))
;; 		   (key  (alist-ref 'servkey  pkt))
;; 		   (pktz (alist-ref 'Z        pkt))
;; 		   (res  (server-ready? uconn host-port key)))
;; 	      (if res
;; 		  res
;; 		  (let* ((pktsdir (get-pkts-dir *toppath*))
;; 			 (pktpath (conc pktsdir"/"pktz".pkt")))
;; 		    (debug:print 0 *default-log-port* "WARNING: pkt with no server "pktpath)
;; 		    (delete-file* pktpath)
;; 		    #f))))
;; 	  serv-pkts))
;; 
;; ;; from viable servers get one that is alive and ready
;; ;;
;; (define (get-the-server uconn apath serv-pkts)
;;   (let loop ((tail serv-pkts))
;;     (if (null? tail)
;; 	#f
;; 	(let* ((spkt  (car tail))
;; 	       (host  (alist-ref 'ipaddr spkt))
;; 	       (port  (alist-ref 'port spkt))
;; 	       (host-port (conc host":"port))
;; 	       (dbpth (alist-ref 'dbpath spkt))
;; 	       (srvkey (alist-ref 'Z spkt)) ;; (alist-ref 'srvkey spkt))
;; 	       (addr  (server-address spkt)))
;; 	  (if (server-ready? uconn host-port srvkey)
;; 	      spkt
;; 	      (loop (cdr tail)))))))
;; 
;; ;; am I the "first" in line server? I.e. my D card is smallest
;; ;; use Z card as tie breaker
;; ;;
;; (define (get-best-candidate serv-pkts dbpath)
;;   (if (null? serv-pkts)
;;       #f
;;       (let loop ((tail serv-pkts)
;; 		 (best  (car serv-pkts)))
;; 	(if (null? tail)
;; 	    best
;; 	    (let* ((candidate (car tail))
;; 		   (candidate-bd (string->number (alist-ref 'D candidate)))
;; 		   (best-bd      (string->number (alist-ref 'D best)))
;; 		   ;; bigger number is younger
;; 		   (candidate-z  (alist-ref 'Z candidate))
;; 		   (best-z       (alist-ref 'Z best))
;; 		   (new-best     (cond
;; 				  ((> best-bd candidate-bd) ;; best is younger than candidate
;; 				   candidate)
;; 				  ((< best-bd candidate-bd) ;; candidate is younger than best
;; 				   best)
;; 				  (else
;; 				   (if (string>=? best-z candidate-z)
;; 				       best
;; 				       candidate))))) ;; use Z card as tie breaker
;; 	      (if (null? tail)
;; 		  new-best
;; 		  (loop (cdr tail) new-best)))))))
;; 	  
;; 
;; ;;======================================================================
;; ;; END NEW SERVER METHOD
;; ;;======================================================================
;; 
;; ;; if .db/main.db check the pkts
;; ;; 
;; (define (rmt:wait-for-server pkts-dir db-file server-key)
;;   (let* ((sdat *db-serv-info*))
;;     (let loop ((start-time (current-seconds))
;; 	       (changed    #t)
;; 	       (last-sdat  "not this"))
;;       (begin ;; let ((sdat #f))
;; 	(thread-sleep! 0.01)
;; 	(debug:print-info 0 *default-log-port* "Waiting for server alive signature")
;; 	(mutex-lock! *heartbeat-mutex*)
;; 	(set! sdat *db-serv-info*)
;; 	(mutex-unlock! *heartbeat-mutex*)
;; 	(if (and sdat
;; 		 (not changed)
;; 		 (> (- (current-seconds) start-time) 2))
;; 	    (let* ((uconn (servdat-uconn sdat)))
;; 	      (servdat-status-set! sdat 'iface-stable)
;; 	      (debug:print-info 0 *default-log-port* "Received server alive signature, now attempting to lock in server")
;; 	      ;; create a server pkt in *toppath*/.meta/srvpkts
;; 	      
;; 	      ;; TODO:
;; 	      ;;   1. change sdat to stuct
;; 	      ;;   2. add uuid to struct
;; 	      ;;   3. update uuid in sdat here
;; 	      ;;
;; 	      (servdat-uuid-set! sdat
;; 				 (register-server
;; 				  pkts-dir *srvpktspec*
;; 				  (get-host-name)
;; 				  (servdat-port sdat) server-key
;; 				  (servdat-host sdat) db-file))
;; 	      ;; (set! *my-signature* (servdat-uuid sdat)) ;; replace with Z, no, stick with proper key
;; 	      ;; now read pkts and see if we are a contender
;; 	      (let* ((all-pkts     (get-all-server-pkts pkts-dir *srvpktspec*))
;; 		     (viables      (get-viable-servers all-pkts db-file))
;; 		     (alive        (remove-pkts-if-not-alive uconn viables))
;; 		     (best-srv     (get-best-candidate alive db-file))
;; 		     (best-srv-key (if best-srv (alist-ref 'servkey best-srv) #f))
;; 		     (i-am-srv     (equal? best-srv-key server-key))
;; 		     (delete-pkt   (lambda ()
;; 				     (let* ((pktfile (conc (get-pkts-dir *toppath*)
;; 							 "/" (servdat-uuid *db-serv-info*)
;; 							 ".pkt")))
;; 				       (debug:print-info 0 *default-log-port* "Attempting to remove bogus pkt file "pktfile)
;; 				       (delete-file* pktfile))))) ;; remove immediately instead of waiting for on-exit
;; 		(debug:print 0 *default-log-port* "best-srv-key: "best-srv-key", server-key: "server-key", i-am-srv: "i-am-srv)
;; 		;; am I the best-srv, compare server-keys to know
;; 		(if i-am-srv
;; 		    (if (get-lock-db sdat db-file (servdat-host sdat)(servdat-port sdat)) ;; (db:get-iam-server-lock *dbstruct-db* *toppath* run-id)
;; 			(begin
;; 			  (debug:print-info 0 *default-log-port* "I'm the server!")
;; 			  (servdat-dbfile-set! sdat db-file)
;; 			  (servdat-status-set! sdat 'db-locked))
;; 			(begin
;; 			  (debug:print-info 0 *default-log-port* "I'm not the server, exiting.")
;; 			  (bdat-time-to-exit-set! *bdat* #t)
;; 			  (delete-pkt)
;; 			  (thread-sleep! 0.2)
;; 			  (exit)))
;; 		    (begin
;; 		      (debug:print-info 0 *default-log-port*
;; 				   "Keys do not match "best-srv-key", "server-key", exiting.")
;; 		      (bdat-time-to-exit-set! *bdat* #t)
;; 		      (delete-pkt)
;; 		      (thread-sleep! 0.2)
;; 		      (exit)))
;; 		sdat))
;; 	    (begin ;; sdat not yet contains server info
;; 	      (debug:print-info 0 *default-log-port* "Still waiting, last-sdat=" last-sdat)
;; 	      (sleep 4)
;; 	      (if (> (- (current-seconds) start-time) 120) ;; been waiting for two minutes
;; 		  (begin
;; 		    (debug:print-error 0 *default-log-port* "transport appears to have died, exiting server")
;; 		    (exit))
;; 		  (loop start-time
;; 			(equal? sdat last-sdat)
;; 			sdat))))))))
;; 
;; (define (rmt:register-server sinfo apath iface port server-key dbname)
;;   (servdat-conns sinfo) ;; just checking types
;;   (rmt:open-main-connection sinfo apath) ;; we need a channel to main.db
;;   (rmt:send-receive-real sinfo apath      ;; params: host port servkey pid ipaddr dbpath
;; 			 (db:run-id->dbname #f)
;; 			 'register-server `(,iface
;; 					    ,port
;; 					    ,server-key
;; 					    ,(current-process-id)
;; 					    ,iface
;; 					    ,apath
;; 					    ,dbname)))
;; 
;; (define (rmt:get-count-servers sinfo apath)
;;   (servdat-conns sinfo) ;; just checking types
;;   (rmt:open-main-connection sinfo apath) ;; we need a channel to main.db
;;   (rmt:send-receive-real sinfo apath      ;; params: host port servkey pid ipaddr dbpath
;; 			 (db:run-id->dbname #f)
;; 			 'get-count-servers `(,apath)))

(define (rmt:get-servers-info apath)
  (rmt:send-receive 'get-servers-info #f `(,apath)))

;; (define (rmt:deregister-server db-serv-info apath iface port server-key dbname)
;;   (rmt:open-main-connection db-serv-info apath) ;; we need a channel to main.db
;;   (rmt:send-receive-real db-serv-info apath      ;; params: host port servkey pid ipaddr dbpath
;;                          (db:run-id->dbname #f)
;;                          'deregister-server `(,iface
;;                                               ,port
;;                                               ,server-key
;;                                               ,(current-process-id)
;;                                               ,iface
;;                                               ,apath
;;                                               ,dbname)))
;; 
;; (define (rmt:wait-for-stable-interface #!optional (num-tries-allowed 100))
;;   ;; wait until *db-serv-info* stops changing
;;   (let* ((stime (current-seconds)))
;;     (let loop ((last-host  #f)
;; 	       (last-port  #f)
;; 	       (tries 0))
;;       (let* ((curr-host (and *db-serv-info* (servdat-host *db-serv-info*)))
;; 	     (curr-port (and *db-serv-info* (servdat-port *db-serv-info*))))
;; 	;; first we verify port and interface, update *db-serv-info* in need be.
;; 	(cond
;; 	 ((> tries num-tries-allowed)
;; 	  (debug:print 0 *default-log-port* "rmt:keep-running, giving up after trying for several minutes.")
;; 	  (exit 1))
;; 	 ((not *db-serv-info*)
;; 	  (thread-sleep! 0.25)
;; 	  (loop curr-host curr-port (+ tries 1)))
;; 	 ((or (not last-host)(not last-port))
;; 	  (debug:print 0 *default-log-port* "rmt:keep-running, still no interface, tries="tries)
;; 	  (thread-sleep! 0.25)
;; 	  (loop curr-host curr-port (+ tries 1)))
;; 	 ((or (not (equal? last-host curr-host))
;; 	      (not (equal? last-port curr-port)))
;; 	  (debug:print-info 0 *default-log-port* "WARNING: interface changed, refreshing iface and port info")
;; 	  (thread-sleep! 0.25)
;; 	  (loop curr-host curr-port (+ tries 1)))
;; 	 ((< (- (current-seconds) stime) 1) ;; keep up the looping until at least 3 seconds have passed
;; 	  (thread-sleep! 0.5)
;; 	  (loop curr-host curr-port (+ tries 1)))
;; 	 (else
;; 	  (rmt:get-signature) ;; sets *my-signature* as side effect
;; 	  (servdat-status-set! *db-serv-info* 'interface-stable)
;; 	  (debug:print 0 *default-log-port*
;; 		       "SERVER STARTED: " curr-host
;; 		       ":" curr-port
;; 		       " AT " (current-seconds) " server signature: " *my-signature*
;; 		       " with "(servdat-trynum *db-serv-info*)" port changes")
;; 	  (flush-output *default-log-port*)
;; 	  #t))))))
;; 
;; ;; run rmt:keep-running in a parallel thread to monitor that the db is being 
;; ;; used and to shutdown after sometime if it is not.
;; ;;
;; (define (rmt:keep-running dbname) 
;;   ;; if none running or if > 20 seconds since 
;;   ;; server last used then start shutdown
;;   ;; This thread waits for the server to come alive
;;   (debug:print-info 0 *default-log-port* "Starting the sync-back, keep alive thread in server")
;; 
;;   (let* ((sinfo             *db-serv-info*)
;; 	 (server-start-time (current-seconds))
;; 	 (pkts-dir          (get-pkts-dir))
;; 	 (server-key        (rmt:get-signature)) ;; This servers key
;; 	 (is-main           (equal? (args:get-arg "-db") ".db/main.db"))
;; 	 (last-access       0)
;; 	 (server-timeout    (server:expiration-timeout))
;; 	 (shutdown-server-sequence (lambda (host port)
;; 				     (set! *unclean-shutdown* #f) ;; Should not be needed anymore
;; 				     (debug:print-info 0 *default-log-port* "Starting to shutdown the server. pid="(current-process-id))
;; 				     ;; (rmt:server-shutdown host port) -- called in on-exit
;; 				     ;; (portlogger:open-run-close portlogger:set-port port "released") called in on-exit
;; 				     (exit)))
;; 	 (timed-out?        (lambda ()
;; 			      (<= (+ last-access server-timeout)
;; 				  (current-seconds)))))
;;     (servdat-dbfile-set! *db-serv-info* (args:get-arg "-db"))
;;     ;; main and run db servers have both got wait logic (could/should merge it)
;;     (if is-main
;; 	(rmt:wait-for-server pkts-dir dbname server-key)
;; 	(rmt:wait-for-stable-interface))
;;     ;; this is our forever loop
;;     (let* ((iface (servdat-host *db-serv-info*))
;; 	   (port  (servdat-port *db-serv-info*))
;; 	   (uconn (servdat-uconn *db-serv-info*)))
;;       (let loop ((count          0)
;; 		 (bad-sync-count 0)
;; 		 (start-time     (current-milliseconds)))
;; 	(if (and (not is-main)
;; 		 (common:low-noise-print 60 "servdat-status"))
;; 	    (debug:print-info 0 *default-log-port* "servdat-status is " (servdat-status *db-serv-info*)))
;; 
;; 	(mutex-lock! *heartbeat-mutex*)
;; 	;; set up the database handle
;; 	(if (not *dbstruct-db*) ;; no db opened yet, open the db and register with main if appropriate
;; 	      (let ((watchdog (bdat-watchdog *bdat*)))
;; 		(debug:print 0 *default-log-port* "SERVER: dbprep")
;; 		(db:setup dbname) ;; sets *dbstruct-db* as side effect
;; 		(servdat-status-set! *db-serv-info* 'db-opened)
;; 		;; IFF I'm not main, call into main and register self
;; 		(if (not is-main)
;; 		    (let ((res (rmt:register-server sinfo
;; 						    *toppath* iface port
;; 						    server-key dbname)))
;; 		      (if res ;; we are the server
;; 			  (servdat-status-set! *db-serv-info* 'have-interface-and-db)
;; 			  ;; now check that the db locker is alive, clear it out if not
;; 			  (let* ((serv-info (rmt:server-info *toppath* dbname)))
;; 			    (match serv-info
;; 				   ((host port servkey pid ipaddr apath dbpath)
;; 				    (if (not (server-ready? uconn (conc host":"port) servkey))
;; 					(begin
;; 					  (debug:print-info 0 *default-log-port* "Server registered but not alive. Removing and trying again.")
;; 					  (rmt:deregister-server sinfo apath host port servkey dbpath) ;; servkey pid ipaddr apath dbpath)
;; 					  (loop (+ count 1) bad-sync-count start-time))))
;; 				   (else
;; 				    (debug:print 0 *default-log-port* "We are not the server for "dbname", exiting. Server info is: "serv-info)
;; 				    (exit)))))))
;; 		(debug:print 0 *default-log-port*
;; 			     "SERVER: running, db "dbname" opened, megatest version: "
;; 			   (common:get-full-version))
;; 	      ;; start the watchdog
;; 
;; 	      ;; is this really needed?
;; 	      
;; 	      #;(if watchdog
;; 		  (if (not (member (thread-state watchdog)
;; 				   '(ready running blocked
;; 					   sleeping dead)))
;; 		      (begin
;; 			(debug:print-info 0 *default-log-port* "Starting watchdog thread (in state "(thread-state watchdog)")")
;; 			(thread-start! watchdog))
;; 		      (debug:print-info 0 *default-log-port* "Not starting watchdog thread (in state "(thread-state watchdog)")"))
;; 		  (debug:print 0 *default-log-port* "ERROR: *watchdog* not setup, cannot start it."))
;; 	      #;(loop (+ count 1) bad-sync-count start-time)
;; 	      ))
;; 	
;; 	(db:sync-inmem->disk *dbstruct-db* *toppath* dbname force-sync: #t)
;; 	
;; 	(mutex-unlock! *heartbeat-mutex*)
;; 	
;; 	;; when things go wrong we don't want to be doing the various
;; 	;; queries too often so we strive to run this stuff only every
;; 	;; four seconds or so.
;; 	(let* ((sync-time (- (current-milliseconds) start-time))
;; 	       (rem-time  (quotient (- 4000 sync-time) 1000)))
;; 	  (if (and (<= rem-time 4)
;; 		   (>  rem-time 0))
;; 	      (thread-sleep! rem-time)))
;;     
;; 	;; Transfer *db-last-access* to last-access to use in checking that we are still alive
;; 	(set! last-access *db-last-access*)
;; 	
;; 	(if (< count 1) ;; 3x3 = 9 secs aprox
;; 	    (loop (+ count 1) bad-sync-count (current-milliseconds)))
;; 	
;; 	(if (common:low-noise-print 60 "dbstats")
;; 	    (begin
;; 	      (debug:print 0 *default-log-port* "Server stats:")
;; 	      (db:print-current-query-stats)))
;; 	(let* ((hrs-since-start  (/ (- (current-seconds) server-start-time) 3600)))
;; 	  (cond
;; 	   ((not *server-run*)
;; 	    (debug:print-info 0 *default-log-port* "*server-run* set to #f. Shutting down.")
;; 	    (shutdown-server-sequence (get-host-name) port))
;; 	   ((timed-out?)
;; 	    (debug:print-info 0 *default-log-port* "Server timed out. seconds since last db access: " (- (current-seconds) last-access))
;; 	    (shutdown-server-sequence (get-host-name) port))
;; 	   ((and *server-run*
;; 		 (or (not (timed-out?))
;; 		     (if is-main ;; do not exit if there are other servers (keep main open until all others gone)
;; 			 (> (rmt:get-count-servers sinfo *toppath*) 1)
;; 			 #f)))
;; 	    (if (common:low-noise-print 120 "server continuing")
;; 		(debug:print-info 0 *default-log-port* "Server continuing, seconds since last db access: " (- (current-seconds) last-access)))
;; 	    (loop 0 bad-sync-count (current-milliseconds)))
;; 	   (else
;; 	    (set! *unclean-shutdown* #f)
;; 	    (debug:print-info 0 *default-log-port* "Server timed out. seconds since last db access: " (- (current-seconds) last-access))
;; 	    (shutdown-server-sequence (get-host-name) port)
;; 	    #;(debug:print-info 0 *default-log-port* "Sending 'quit to server, received: "
;; 			      (open-send-receive-nn (conc iface":"port)      ;; do this here and not in server-shutdown
;; 						    (sexpr->string 'quit))))))))))
;; 
;; (define (rmt:get-reasonable-hostname)
;;   (let* ((inhost (or (args:get-arg "-server") "-")))
;;     (if (equal? inhost "-")
;; 	(get-host-name)
;; 	inhost)))
;; 
;; Call this to start the actual server
;;
;; all routes though here end in exit ...
;;
;; This is the point at which servers are started
;;
(define (rmt:server-launch dbname)
  (debug:print-info 0 *default-log-port* "Entered rmt:server-launch")
  #;(let* ((th2 (make-thread (lambda ()
			     (debug:print-info 0 *default-log-port* "Server run thread started")
			     (rmt:run (rmt:get-reasonable-hostname)))
			   "Server run"))
	 (th3 (make-thread (lambda ()
			     (debug:print-info 0 *default-log-port* "Server monitor thread started")
			     (if (args:get-arg "-server")
				 (rmt:keep-running dbname)))
			     "Keep running")))
    (thread-start! th2)
    (thread-sleep! 0.252) ;; give the server time to settle before starting the keep-running monitor.
    (thread-start! th3)
    (set! *didsomething* #t)
    (thread-join! th2)
    (thread-join! th3))
  #f)
;; 	    
;; ;;======================================================================
;; ;;  S E R V E R   -  D I R E C T   C A L L S
;; ;;======================================================================
;; 
;; (define (rmt:kill-server run-id)
;;   (rmt:send-receive 'kill-server #f (list run-id)))
;; 
;; (define (rmt:start-server run-id)
;;   (rmt:send-receive 'start-server #f (list run-id)))
;; 
;; (define (rmt:server-info apath dbname)
;;   (rmt:send-receive 'get-server-info #f (list apath dbname)))
;; 
;; ;;======================================================================
;; ;; Nanomsg transport
;; ;;======================================================================
;; 
;; #;(define (is-port-in-use port-num)
;;   (let* ((ret #f))
;;     (let-values (((inp oup pid)
;; 		  (process "netstat" (list  "-tulpn" ))))
;;       (let loop ((inl (read-line inp)))
;;         (if (not (eof-object? inl))
;;             (begin 
;; 	      (if (string-search (regexp (conc ":" port-num)) inl)
;; 		  (begin
;; 					;(print "Output: "  inl)
;; 		    (set! ret  #t))
;; 		  (loop (read-line inp)))))))
;;     ret))
;; 
;; #;(define (open-nn-connection host-port)
;;   (let ((req  (make-req-socket))
;;         (uri  (conc "tcp://" host-port)))
;;     (nng-dial req uri)
;;     (socket-set! req 'nng/recvtimeo 2000)
;;     req))
;; 
;; #;(define (send-receive-nn req msg)
;;   (nng-send req msg)
;;   (nng-recv req))
;; 
;; #;(define (close-nn-connection req)
;;   (nng-close! req))
;;   
;; ;; ;; open connection to server, send message, close connection
;; ;; ;;
;; ;; (define (open-send-close-nn host-port msg #!key (timeout 3) ) ;; default timeout is 3 seconds
;; ;;   (let ((req  (make-req-socket 'req))
;; ;;         (uri  (conc "tcp://" host-port))
;; ;;         (res  #f)
;; ;;         ;; (contacts (alist-ref 'contact attrib))
;; ;;         ;; (mode (alist-ref 'mode attrib))
;; ;; 	)
;; ;;     (socket-set! req 'nng/recvtimeo 2000)
;; ;;     (handle-exceptions
;; ;;      exn
;; ;;      (let ((emsg ((condition-property-accessor 'exn 'message) exn)))
;; ;;        ;; Send notification       
;; ;;        (debug:print 0 *default-log-port* "ERROR: Failed to connect / send to " uri " message was \"" emsg "\"" )
;; ;;        #f)
;; ;;      (nng-dial req uri)
;; ;;      ;; (print "Connected to the server " )
;; ;;      (nng-send req msg)
;; ;;      ;; (print "Request Sent")  
;; ;;      (let* ((th1  (make-thread (lambda ()
;; ;;                                  (let ((resp (nng-recv req)))
;; ;;                                    (nng-close! req)
;; ;;                                    (set! res (if (equal? resp "ok")
;; ;;                                                  #t
;; ;;                                                  #f))))
;; ;;                                "recv thread"))
;; ;;             (th2 (make-thread (lambda ()
;; ;;                                 (thread-sleep! timeout)
;; ;;                                 (thread-terminate! th1))
;; ;; 			      "timer thread")))
;; ;;        (thread-start! th1)
;; ;;        (thread-start! th2)
;; ;;        (thread-join! th1)
;; ;;        res))))
;; ;; 
;; #;(define (open-send-receive-nn host-port msg #!key (timeout 3) ) ;; default timeout is 3 seconds
;;   (let ((req  (make-req-socket))
;;         (uri  (conc "tcp://" host-port))
;;         (res  #f)) 
;;     (handle-exceptions
;;      exn
;;      (let ((emsg ((condition-property-accessor 'exn 'message) exn)))
;;        ;; Send notification      
;;        (debug:print 0 *default-log-port* "ERROR: Failed to connect / send to " uri " message was \"" emsg "\", exn=" exn)
;;        #f)
;;      (nng-dial req uri)
;;      (nng-send req msg)
;;      (let* ((th1  (make-thread (lambda ()
;;                                  (let ((resp (nng-recv req)))
;;                                    (nng-close! req)
;;                                    ;; (print resp)
;;                                    (set! res resp)))
;;                                "recv thread"))
;;             (th2 (make-thread (lambda ()
;;                                 (thread-sleep! timeout)
;;                                 (thread-terminate! th1))
;;                              "timer thread")))
;;        (thread-start! th1)
;;        (thread-start! th2)
;;        (thread-join! th1)
;;        res))))
;; 
;; ;;======================================================================
;; ;; S E R V E R   U T I L I T I E S 
;; ;;======================================================================
;; 
;; ;; run ping in separate process, safest way in some cases
;; ;;
;; #;(define (server:ping-server ifaceport)
;;   (with-input-from-pipe 
;;    (conc (common:get-megatest-exe) " -ping " ifaceport)
;;    (lambda ()
;;      (let loop ((inl (read-line))
;; 		(res "NOREPLY"))
;;        (if (eof-object? inl)
;; 	   (case (string->symbol res)
;; 	     ((NOREPLY)  #f)
;; 	     ((LOGIN_OK) #t)
;; 	     (else       #f))
;; 	   (loop (read-line) inl))))))
;; 
;; ;; NOT USED (well, ok, reference in rpc-transport but otherwise not used).
;; ;;
;; #;(define (server:login toppath)
;;   (lambda (toppath)
;;     (set! *db-last-access* (current-seconds)) ;; might not be needed.
;;     (if (equal? *toppath* toppath)
;; 	#t
;; 	#f)))
;; 
;; ;; (define server:sync-lock-token "SERVER_SYNC_LOCK")
;; ;; (define (server:release-sync-lock)
;; ;;   (db:no-sync-del! *no-sync-db* server:sync-lock-token))
;; ;; (define (server:have-sync-lock?)
;; ;;   (let* ((have-lock-pair (db:no-sync-get-lock *no-sync-db* server:sync-lock-token))
;; ;;          (have-lock?     (car have-lock-pair))
;; ;;          (lock-time      (cdr have-lock-pair))
;; ;;          (lock-age       (- (current-seconds) lock-time)))
;; ;;     (cond
;; ;;      (have-lock? #t)
;; ;;      ((>lock-age
;; ;;        (* 3 (configf:lookup-number *configdat* "server" "minimum-intersync-delay" default: 180)))
;; ;;       (server:release-sync-lock)
;; ;;       (server:have-sync-lock?))
;; ;;      (else #f))))

)

Added ulex-none/ulex.scm version [4551982255].



















































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
;; ulex: Distributed sqlite3 db
;;;
;; Copyright (C) 2018-2021 Matt Welland
;; Redistribution and use in source and binary forms, with or without
;; modification, is permitted.
;;
;; THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS
;; OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
;; WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
;; ARE DISCLAIMED.  IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE
;; LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
;; CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT
;; OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
;; BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
;; LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
;; (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
;; USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH
;; DAMAGE.

;;======================================================================
;; ABOUT:
;;   See README in the distribution at https://www.kiatoa.com/fossils/ulex
;; NOTES:
;;   Why sql-de-lite and not say, dbi?  - performance mostly, then simplicity.
;;
;;======================================================================

(module ulex
	*
	#;(
     
     ;; NOTE: looking for the handler proc - find the run-listener :)
     
     run-listener     ;; (run-listener handler-proc [port]) => uconn

     ;; NOTE: handler-proc params;
     ;;       (handler-proc rem-host-port qrykey cmd params)

     send-receive     ;; (send-receive uconn host-port cmd data)

     ;; NOTE: cmd can be any plain text symbol except for these;
     ;;         'ping 'ack 'goodbye 'response
     
     set-work-handler ;; (set-work-handler proc)

     wait-and-close   ;; (wait-and-close uconn)

     ulex-listener?
     
     ;; needed to get the interface:port that was automatically found
     udat-port
     udat-host-port
     
     ;; for testing only
     ;; pp-uconn
     
     ;; parameters
     work-method   ;; parameter; 'threads, 'mailbox, 'limited, 'direct
     return-method ;; parameter; 'mailbox, 'polling, 'direct
     )

(import scheme
	chicken.base
	chicken.file
	chicken.io
	chicken.time
	chicken.condition
	chicken.string
	chicken.sort
	chicken.pretty-print
	
	address-info
	mailbox
	matchable
	;; queues
	regex
	regex-case
	simple-exceptions
	s11n
	srfi-1
	srfi-18
	srfi-4
	srfi-69
	system-information
	tcp6
	typed-records
	)

;; ;; udat struct, used by both caller and callee
;; ;; instantiated as uconn by convention
;; ;;
;; (defstruct udat
;;   ;; the listener side
;;   (port #f)
;;   (host-port #f)
;;   (socket #f)
;;   ;; the peers
;;   (peers  (make-hash-table)) ;; host:port->peer
;;   ;; work handling
;;   (work-queue (make-mailbox))
;;   (work-proc  #f)                ;; set by user
;;   (cnum       0)                 ;; cookie number
;;   (mboxes     (make-hash-table)) ;; for the replies
;;   (avail-cmboxes '())            ;; list of (<cookie> . <mbox>) for re-use
;;   ;; threads
;;   (numthreads 10)
;;   (cmd-thread #f)
;;   (work-queue-thread #f)
;;   (num-threads-running 0)
;;   ) 
;; 
;; ;; Parameters
;; 
;; ;; work-method:
;; (define work-method (make-parameter 'mailbox))
;; ;;    mailbox - all rdat goes through mailbox
;; ;;    threads - all rdat immediately executed in new thread
;; ;;    direct  - no queuing
;; ;;
;; 
;; ;; return-method, return the result to waiting send-receive:
;; (define return-method (make-parameter 'mailbox))
;; ;;    mailbox - create a mailbox and use it for passing returning results to send-receive
;; ;;    polling - put the result in a hash table keyed by qrykey and send-receive can poll it for result
;; ;;    direct  - no queuing, result is passed back in single tcp connection
;; ;;
;; 
;; ;; ;; struct for keeping track of others we are talking to
;; ;; ;;
;; ;; (defstruct pdat
;; ;;   (host-port  #f)
;; ;;   (conns      '()) ;; list of pcon structs, pop one off when calling the peer
;; ;;   )
;; ;; 
;; ;; ;; struct for peer connections, keep track of expiration etc.
;; ;; ;;
;; ;; (defstruct pcon
;; ;;   (inp #f)
;; ;;   (oup #f)
;; ;;   (exp (+ (current-seconds) 59)) ;; expires at this time, set to (+ (current-seconds) 59)
;; ;;   (lifetime (+ (current-seconds) 600)) ;; throw away and create new after five minutes
;; ;;   )
;; 
;; ;;======================================================================
;; ;; listener
;; ;;======================================================================
;; 
;; ;; is uconn a ulex connector (listener)
;; ;;
;; (define (ulex-listener? uconn)
;;   (udat? uconn))
;; 
;; ;; create a tcp listener and return a populated udat struct with
;; ;; my port, address, hostname, pid etc.
;; ;; return #f if fail to find a port to allocate.
;; ;;
;; ;;  if udata-in is #f create the record
;; ;;  if there is already a serv-listener return the udata
;; ;;
;; (define (setup-listener uconn #!optional (port 4242))
;;   (handle-exceptions
;;    exn
;;    (if (< port 65535)
;;        (setup-listener uconn (+ port 1))
;;        #f)
;;    (connect-listener uconn port)))
;; 
;; (define (connect-listener uconn port)
;;   ;; (tcp-listener-socket LISTENER)(socket-name so)
;;   ;; sockaddr-address, sockaddr-port, sockaddr->string
;;   (let* ((tlsn (tcp-listen port 1000 #f)) ;; (tcp-listen TCPPORT [BACKLOG [HOST]])
;; 	 (addr (get-my-best-address))) ;; (hostinfo-addresses (host-information (current-hostname)))
;;     (udat-port-set!      uconn port)
;;     (udat-host-port-set! uconn (conc addr":"port))
;;     (udat-socket-set!    uconn tlsn)
;;     uconn))
;; 
;; ;; run-listener does all the work of starting a listener in a thread
;; ;; it then returns control
;; ;;
;; (define (run-listener handler-proc #!optional (port-suggestion 4242))
;;   (let* ((uconn (make-udat)))
;;     (udat-work-proc-set! uconn handler-proc)
;;     (if (setup-listener uconn port-suggestion)
;; 	(let* ((th1 (make-thread (lambda ()(ulex-cmd-loop uconn)) "Ulex command loop"))
;; 	       (th2 (make-thread (lambda ()
;; 				   (case (work-method)
;; 				     ((mailbox limited)
;; 				      (process-work-queue uconn))))
;; 				 "Ulex work queue processor")))
;; 	  ;; (tcp-buffer-size 2048)
;; 	  (thread-start! th1)
;; 	  (thread-start! th2)
;; 	  (udat-cmd-thread-set! uconn th1)
;; 	  (udat-work-queue-thread-set! uconn th2)
;; 	  (print "cmd loop and process workers started, listening on "(udat-host-port uconn)".")
;; 	  uconn)
;; 	(assert #f "ERROR: run-listener called without proper setup."))))
;; 
;; (define (wait-and-close uconn)
;;   (thread-join! (udat-cmd-thread uconn))
;;   (tcp-close (udat-socket uconn)))
;; 
;; ;;======================================================================
;; ;; peers and connections
;; ;;======================================================================
;; 
;; (define *send-mutex* (make-mutex))
;; 
;; ;; send structured data to recipient
;; ;;
;; ;;  NOTE: qrykey is what was called the "cookie" previously
;; ;;
;; ;;     retval tells send to expect and wait for return data (one line) and return it or time out
;; ;;       this is for ping where we don't want to necessarily have set up our own server yet.
;; ;;
;; ;; NOTE: see below for beginnings of code to allow re-use of tcp connections
;; ;;        - I believe (without substantial evidence) that re-using connections will
;; ;;          be beneficial ...
;; ;;
;; (define (send udata host-port qrykey cmd params)
;;   (let* ((my-host-port (udat-host-port udata))          ;; remote will return to this
;; 	 (isme         #f #;(equal? host-port my-host-port)) ;; calling myself?
;; 	 ;; dat is a self-contained work block that can be sent or handled locally
;; 	 (dat          (list my-host-port qrykey cmd params #;(cons (current-seconds)(current-milliseconds)))))
;;     (cond
;;      (isme (ulex-handler udata dat)) ;; no transmission needed
;;      (else
;;       (handle-exceptions ;; TODO - MAKE THIS EXCEPTION CMD SPECIFIC?
;; 	  exn
;; 	  (message exn)
;; 	(begin
;; 	  ;; (mutex-lock! *send-mutex*) ;; DOESN'T SEEM TO HELP
;; 	  (let-values (((inp oup)(tcp-connect host-port)))
;; 	    (let ((res (if (and inp oup)
;; 			   (begin
;; 			     (serialize dat oup)
;; 			     (close-output-port oup)
;; 			     (deserialize inp)
;; 			     )
;; 			   (begin
;; 			     (print "ERROR: send called but no receiver has been setup. Please call setup first!")
;; 			     #f))))
;; 	      (close-input-port inp)
;; 	      ;; (mutex-unlock! *send-mutex*) ;; DOESN'T SEEM TO HELP
;; 	      res)))))))) ;; res will always be 'ack unless return-method is direct
;; 
;; (define (send-via-polling uconn host-port cmd data)
;;   (let* ((qrykey (make-cookie uconn))
;; 	 (sres   (send uconn host-port qrykey cmd data)))
;;     (case sres
;;       ((ack)
;;        (let loop ((start-time (current-milliseconds)))
;; 	 (if (> (current-milliseconds)(+ start-time 10000)) ;; ten seconds timeout
;; 	     (begin
;; 	       (print "ULEX ERROR: timed out waiting for response from "host-port", "cmd" "data)
;; 	       #f)
;; 	     (let* ((result (hash-table-ref/default (udat-mboxes uconn) qrykey #f))) ;; NOTE: we are re-using mboxes hash
;; 	       (if result ;; result is '(status . result-data) or #f for nothing yet
;; 		   (begin
;; 		     (hash-table-delete! (udat-mboxes uconn) qrykey)
;; 		     (cdr result))
;; 		   (begin
;; 		     (thread-sleep! 0.01)
;; 		     (loop start-time)))))))
;;       (else
;;        (print "ULEX ERROR: Communication failed? sres="sres)
;;        #f))))
;; 
;; (define (send-via-mailbox uconn host-port cmd data)
;;   (let* ((cmbox     (get-cmbox uconn)) ;; would it be better to keep a stack of mboxes to reuse?
;; 	 (qrykey    (car cmbox))
;; 	 (mbox      (cdr cmbox))
;; 	 (mbox-time (current-milliseconds))
;; 	 (sres      (send uconn host-port qrykey cmd data))) ;; short res
;;     (if (eq? sres 'ack)
;; 	(let* ((mbox-timeout-secs    120 #;(if (eq? 'primordial (thread-name (current-thread)))
;; 				     #f
;; 				     120)) ;; timeout)
;; 	       (mbox-timeout-result 'MBOX_TIMEOUT)
;; 	       (res                  (mailbox-receive! mbox mbox-timeout-secs mbox-timeout-result))
;; 	       (mbox-receive-time    (current-milliseconds)))
;; 	  ;; (put-cmbox uconn cmbox) ;; reuse mbox and cookie. is it worth it?
;; 	  (hash-table-delete! (udat-mboxes uconn) qrykey)
;; 	  (if (eq? res 'MBOX_TIMEOUT)
;; 	      (begin
;; 		(print "WARNING: mbox timed out for query "cmd", with data "data
;; 		       ", waiting for response from "host-port".")
;; 
;; 		;; here it might make sense to clean up connection records and force clean start?
;; 		;; NO. The progam using ulex needs to do the reset. Right thing here is exception
;; 		
;; 		#f)  ;; convert to raising exception?
;; 	      res))
;; 	(begin
;; 	  (print "ERROR: Communication failed? Got "sres)
;; 	  #f))))
;;   
;; ;; send a request to the given host-port and register a mailbox in udata
;; ;; wait for the mailbox data and return it
;; ;;
;; (define (send-receive uconn host-port cmd data)
;;   (let* ((start-time (current-milliseconds))
;; 	 (result     (cond
;; 		      ((member cmd '(ping goodbye)) ;; these are immediate
;; 		       (send uconn host-port 'ping cmd data))
;; 		      ((eq? (work-method) 'direct)
;; 		       ;; the result from send will be the actual result, not an 'ack
;; 		       (send uconn host-port 'direct cmd data))
;; 		      (else
;; 		       (case (return-method)
;; 			 ((polling)
;; 			  (send-via-polling uconn host-port cmd data))
;; 			 ((mailbox) 
;; 			  (send-via-mailbox uconn host-port cmd data))
;; 			 (else
;; 			  (print "ULEX ERROR: unrecognised return-method "(return-method)".")
;; 			  #f)))))
;; 	 (duration    (- (current-milliseconds) start-time)))
;;     ;; this is ONLY for development and debugging. It will be removed once Ulex is stable.
;;     (if (< 5000 duration)
;; 	(print "ULEX WARNING: round-trip took "(inexact->exact (round (/ duration 1000)))
;; 	       " seconds; "cmd", host-port="host-port", data="data))
;;     result))
;;     
;; 
;; ;;======================================================================
;; ;; responder side
;; ;;======================================================================
;; 
;; ;; take a request, rdat, and if not immediate put it in the work queue
;; ;;
;; ;; Reserved cmds; ack ping goodbye response
;; ;;
;; (define (ulex-handler uconn rdat)
;;   (assert (list? rdat) "FATAL: ulex-handler give rdat as not list")
;;   (match rdat ;;  (string-split controldat)
;;     ((rem-host-port qrykey cmd params);; timedata)
;;      ;; (print "ulex-handler got: "rem-host-port" qrykey: "qrykey" cmd: "cmd" params: "params)
;;      (case cmd
;;        ;; ((ack )(print "Got ack! But why? Should NOT get here.") 'ack)
;;        ((ping)
;; 	;; (print "Got Ping!")
;; 	;; (add-to-work-queue uconn rdat)
;; 	'ack)
;;        ((goodbye)
;; 	;; just clear out references to the caller. NOT COMPLETE
;; 	(add-to-work-queue uconn rdat)
;; 	'ack)
;;        ((response) ;; this is a result from remote processing, send it as mail ...
;; 	(case (return-method)
;; 	  ((polling)
;; 	   (hash-table-set! (udat-mboxes uconn) qrykey (cons 'ok params))
;; 	   'ack)
;; 	  ((mailbox)
;; 	   (let ((mbox (hash-table-ref/default (udat-mboxes uconn) qrykey #f)))
;; 	     (if mbox
;; 		 (begin
;; 		   (mailbox-send! mbox params) ;; params here is our result
;; 		   'ack)
;; 		 (begin
;; 		   (print "ERROR: received result but no associated mbox for cookie "qrykey)
;; 		   'no-mbox-found))))
;; 	  (else (print "ULEX ERROR: unrecognised return-method "(return-method))
;; 		'bad-return-method)))
;;        (else ;; generic request - hand it to the work queue
;; 	(add-to-work-queue uconn rdat)
;; 	'ack)))
;;     (else
;;      (print "ULEX ERROR: bad rdat "rdat)
;;      'bad-rdat)))
;; 
;; ;; given an already set up uconn start the cmd-loop
;; ;;
;; (define (ulex-cmd-loop uconn)
;;   (let* ((serv-listener (udat-socket uconn))
;; 	 (listener      (lambda ()
;; 			  (let loop ((state 'start))
;; 			    (let-values (((inp oup)(tcp-accept serv-listener)))
;; 			      ;; (mutex-lock! *send-mutex*) ;; DOESN'T SEEM TO HELP
;; 			      (let* ((rdat  (deserialize inp)) ;; '(my-host-port qrykey cmd params)
;; 				     (resp  (ulex-handler uconn rdat)))
;; 				(serialize resp oup)
;; 				(close-input-port inp)
;; 				(close-output-port oup)
;; 				;; (mutex-unlock! *send-mutex*) ;; DOESN'T SEEM TO HELP
;; 				)
;; 			      (loop state))))))
;;     ;; start N of them
;;     (let loop ((thnum   0)
;; 	       (threads '()))
;;       (if (< thnum 100)
;; 	  (let* ((th (make-thread listener (conc "listener" thnum))))
;; 	    (thread-start! th)
;; 	    (loop (+ thnum 1)
;; 		  (cons th threads)))
;; 	  (map thread-join! threads)))))
;; 
;; ;; add a proc to the cmd list, these are done symetrically (i.e. in all instances)
;; ;; so that the proc can be dereferenced remotely
;; ;;
;; (define (set-work-handler uconn proc)
;;   (udat-work-proc-set! uconn proc))
;; 
;; ;;======================================================================
;; ;; work queues - this is all happening on the listener side
;; ;;======================================================================
;; 
;; ;; rdat is (rem-host-port qrykey cmd params)
;; 					     
;; (define (add-to-work-queue uconn rdat)
;;   #;(queue-add! (udat-work-queue uconn) rdat)
;;   (case (work-method)
;;     ((threads)
;;      (thread-start! (make-thread (lambda ()
;; 				   (do-work uconn rdat))
;; 				 "worker thread")))
;;     ((mailbox)
;;      (mailbox-send! (udat-work-queue uconn) rdat))
;;     ((direct)
;;      (do-work uconn rdat))
;;     (else
;;      (print "ULEX ERROR: work-method "(work-method)" not recognised, using mailbox.")
;;      (mailbox-send! (udat-work-queue uconn) rdat))))
;;      
;; ;; move the logic to return the result somewhere else?
;; ;;
;; (define (do-work uconn rdat)
;;   (let* ((proc (udat-work-proc uconn))) ;; get it each time - conceivebly it could change
;;     ;; put this following into a do-work procedure
;;     (match rdat
;;       ((rem-host-port qrykey cmd params)
;;        (let* ((start-time (current-milliseconds))
;; 	      (result (proc rem-host-port qrykey cmd params))
;; 	      (end-time (current-milliseconds))
;; 	      (run-time (- end-time start-time)))
;; 	 (case (work-method)
;; 	   ((direct) result)
;; 	   (else
;; 	    (print "ULEX: work "cmd", "params" done in "run-time" ms")
;; 	    ;; send 'response as cmd and result as params
;; 	    (send uconn rem-host-port qrykey 'response result) ;; could check for ack
;; 	    (print "ULEX: response sent back to "rem-host-port" in "(- (current-milliseconds) end-time))))))
;;       (MBOX_TIMEOUT 'do-work-timeout)
;;       (else
;;        (print "ERROR: rdat "rdat", did not match rem-host-port qrykey cmd params")))))
;; 
;; ;; NEW APPROACH:
;; ;;   
;; (define (process-work-queue uconn) 
;;   (let ((wqueue (udat-work-queue uconn))
;; 	(proc   (udat-work-proc  uconn))
;; 	(numthr (udat-numthreads uconn)))
;;     (let loop ((thnum    1)
;; 	       (threads '()))
;;       (let ((thlst (cons (make-thread (lambda ()
;; 					(let work-loop ()
;; 					  (let ((rdat (mailbox-receive! wqueue 24000 'MBOX_TIMEOUT)))
;; 					    (do-work uconn rdat))
;; 					  (work-loop)))
;; 				      (conc "work thread " thnum))
;; 			 threads)))
;; 	(if (< thnum numthr)
;; 	    (loop (+ thnum 1)
;; 		  thlst)
;; 	    (begin
;; 	      (print "ULEX: Starting "(length thlst)" worker threads.")
;; 	      (map thread-start! thlst)
;; 	      (print "ULEX: Threads started. Joining all.")
;; 	      (map thread-join! thlst)))))))
;; 
;; ;; below was to enable re-use of connections. This seems non-trivial so for
;; ;; now lets open on each call
;; ;;
;; ;; ;; given host-port get or create peer struct
;; ;; ;;
;; ;; (define (udat-get-peer uconn host-port)
;; ;;   (or (hash-table-ref/default (udat-peers uconn) host-port #f)
;; ;;       ;; no peer, so create pdat and init it
;; ;;       
;; ;;       ;; NEED stack of connections, pop and use; inp, oup,
;; ;;       ;; creation_time (remove and create new if over 24hrs old
;; ;;       ;; 
;; ;;       (let ((pdat (make-pdat host-port: host-port)))
;; ;; 	(hash-table-set! (udat-peers uconn) host-port pdat)
;; ;; 	pdat)))
;; ;; 
;; ;; ;; is pcon alive
;; ;; 
;; ;; ;; given host-port and pdat get a pcon
;; ;; ;;
;; ;; (define (pdat-get-pcon pdat host-port)
;; ;;   (let loop ((conns (pdat-conns pdat)))
;; ;;     (if (null? conns) ;; none? make and return - do NOT add - it will be pushed back on list later
;; ;; 	(init-pcon (make-pcon))
;; ;; 	(let* ((conn (pop conns)))
;; ;; 	  
;; ;; ;; given host-port get a pcon struct
;; ;; ;;
;; ;; (define (udat-get-pcon 
;;       
;; ;;======================================================================
;; ;; misc utils
;; ;;======================================================================
;; 
;; (define (make-cookie uconn)
;;   (let ((newcnum (+ (udat-cnum uconn) 1)))
;;     (udat-cnum-set! uconn newcnum)
;;     (conc (udat-host-port uconn) ":"
;; 	  newcnum)))
;; 
;; ;; cookie/mboxes
;; 
;; ;; we store each mbox with a cookie (<cookie> . <mbox>)
;; ;;
;; (define (get-cmbox uconn)
;;   (if (null? (udat-avail-cmboxes uconn))
;;       (let ((cookie (make-cookie uconn))
;; 	    (mbox   (make-mailbox)))
;; 	(hash-table-set! (udat-mboxes uconn) cookie mbox)
;; 	`(,cookie . ,mbox))
;;       (let ((cmbox (car (udat-avail-cmboxes uconn))))
;; 	(udat-avail-cmboxes-set! uconn (cdr (udat-avail-cmboxes uconn)))
;; 	cmbox)))
;; 
;; (define (put-cmbox uconn cmbox)
;;   (udat-avail-cmboxes-set! uconn (cons cmbox (udat-avail-cmboxes uconn))))
;; 
;; (define (pp-uconn uconn)
;;   (pp (udat->alist uconn)))
;; 
;;   
;; ;;======================================================================
;; ;; network utilities
;; ;;======================================================================
;; 
;; ;; NOTE: Look at address-info egg as alternative to some of this
;; 
;; (define (rate-ip ipaddr)
;;   (regex-case ipaddr
;;     ( "^127\\..*" _ 0 )
;;     ( "^(10\\.0|192\\.168)\\..*" _ 1 )
;;     ( else 2 ) ))
;; 
;; ;; Change this to bias for addresses with a reasonable broadcast value?
;; ;;
;; (define (ip-pref-less? a b)
;;   (> (rate-ip a) (rate-ip b)))
;; 
;; (define (get-my-best-address)
;;   (let ((all-my-addresses (get-all-ips)))
;;     (cond
;;      ((null? all-my-addresses)
;;       (get-host-name))                                          ;; no interfaces?
;;      ((eq? (length all-my-addresses) 1)
;;       (car all-my-addresses))                      ;; only one to choose from, just go with it
;;      (else
;;       (car (sort all-my-addresses ip-pref-less?))))))
;; 
;; (define (get-all-ips-sorted)
;;   (sort (get-all-ips) ip-pref-less?))
;; 
;; (define (get-all-ips)
;;   (map address-info-host
;;        (filter (lambda (x)
;; 		 (equal? (address-info-type x) "tcp"))
;; 	       (address-infos (get-host-name)))))
;; 
)

Modified ulex-simple/dbmgr.scm from [cc60b18701] to [639e4c49a8].

156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
;; connections for other servers happens by requesting from main
;;
;; TODO: This is unnecessarily re-creating the record in the hash table
;;
(define (rmt:open-main-connection remdat apath)
  (let* ((fullpath (db:dbname->path apath ".db/main.db"))
	 (conns    (servdat-conns remdat))
	 (conn     (rmt:get-conn remdat apath ".db/main.db")) ;; (hash-table-ref/default conns fullpath #f)) ;; TODO - create call for this
	 (start-rmt:run (lambda ()
			  (let* ((th1 (make-thread (lambda ()(rmt:run (get-host-name))) "non-db mode server")))
			    (thread-start! th1)
			    (thread-sleep! 1)
			    (let loop ((count 0))
			      (assert (< count 30) "FATAL: responder failed to initialize in rmt:open-main-connection")
			      (if (or (not *db-serv-info*)
				      (not (servdat-uconn *db-serv-info*)))
				  (begin
				    (thread-sleep! 1)
				    (loop (+ count 1)))
				  (begin
				    (servdat-mode-set! *db-serv-info* 'non-db)
				    (servdat-uconn *db-serv-info*)))))))
	 (myconn    (servdat-uconn *db-serv-info*)))
    (cond
     ((not myconn)
      (start-rmt:run)
      (rmt:open-main-connection remdat apath))
     ((and conn                                             ;; conn is NOT a socket, just saying ...
	   (< (current-seconds) (conndat-expires conn)))
      #t) ;; we are current and good to go - we'll deal elsewhere with a server that was killed or died 
     ((and conn
	   (>= (current-seconds)(conndat-expires conn)))
      (debug:print-info 0 *default-log-port* "connection to "fullpath" server expired. Reconnecting.")
      (rmt:drop-conn remdat apath ".db/main.db") ;;







|
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<

<
<
<







156
157
158
159
160
161
162
163















164



165
166
167
168
169
170
171
;; connections for other servers happens by requesting from main
;;
;; TODO: This is unnecessarily re-creating the record in the hash table
;;
(define (rmt:open-main-connection remdat apath)
  (let* ((fullpath (db:dbname->path apath ".db/main.db"))
	 (conns    (servdat-conns remdat))
	 (conn     (rmt:get-conn remdat apath ".db/main.db"))) ;; (hash-table-ref/default conns fullpath #f)) ;; TODO - create call for this















    (cond



     ((and conn                                             ;; conn is NOT a socket, just saying ...
	   (< (current-seconds) (conndat-expires conn)))
      #t) ;; we are current and good to go - we'll deal elsewhere with a server that was killed or died 
     ((and conn
	   (>= (current-seconds)(conndat-expires conn)))
      (debug:print-info 0 *default-log-port* "connection to "fullpath" server expired. Reconnecting.")
      (rmt:drop-conn remdat apath ".db/main.db") ;;
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
    (if *localmode*
	(api:execute-requests *dbstruct* cmd params)
	(begin
	  (rmt:open-main-connection sinfo apath)
	  (if rid (rmt:general-open-connection sinfo apath dbname))
	  #;(if (not (member cmd '(log-to-main)))
	      (debug:print-info 0 *default-log-port* "rmt:send-receive "cmd" params="params))
	  (rmt:send-receive-real sinfo apath dbname cmd params)))))

;; db is at apath/.db/dbname, rid is an intermediary solution and will be removed
;; sometime in the future
;;
(define (rmt:send-receive-real sinfo apath dbname cmd params)
  (assert (not (eq? 'primordial (thread-name (current-thread)))) "FATAL: Do not call rmt:send-receive-real in the primodial thread.")
  (let* ((cdat (rmt:get-conn sinfo apath dbname)))
    (assert cdat "FATAL: rmt:send-receive-real called without the needed channels opened")
    (let* ((uconn    (servdat-uconn sinfo)) ;; get the interface to ulex
	   ;; then send-receive using the ulex layer to host-port stored in cdat
	   (res      (send-receive uconn (conndat-hostport cdat) cmd params))
	   #;(th1      (make-thread (lambda ()
				    (set! res (send-receive uconn (conndat-hostport cdat) cmd params)))
				  "send-receive thread")))
      ;; (thread-start! th1)
      ;; (thread-join! th1)   ;; gratuitious thread stuff is so that mailbox is not used in primordial thead
      ;; since we accessed the server we can bump the expires time up
      (conndat-expires-set! cdat (+ (current-seconds)
				    (server:expiration-timeout)
				    -2)) ;; two second margin for network time misalignments etc.
      res)))

;; db is at apath/.db/dbname, rid is an intermediary solution and will be removed
;; sometime in the future.
;;
;; Purpose - call the main.db server and request a server be started
;; for the given area path and dbname
;;







<
<
<
<
<
<
|
|
|
|
|
|
<
<
<
<
<
<
|
|
|
|







294
295
296
297
298
299
300






301
302
303
304
305
306






307
308
309
310
311
312
313
314
315
316
317
    (if *localmode*
	(api:execute-requests *dbstruct* cmd params)
	(begin
	  (rmt:open-main-connection sinfo apath)
	  (if rid (rmt:general-open-connection sinfo apath dbname))
	  #;(if (not (member cmd '(log-to-main)))
	      (debug:print-info 0 *default-log-port* "rmt:send-receive "cmd" params="params))






	  (assert (not (eq? 'primordial (thread-name (current-thread)))) "FATAL: Do not call rmt:send-receive-real in the primodial thread.")
	  (let* ((cdat (rmt:get-conn sinfo apath dbname)))
	    (assert cdat "FATAL: rmt:send-receive-real called without the needed channels opened")
	    (let* ((uconn    (servdat-uconn sinfo)) ;; get the interface to ulex
		   ;; then send-receive using the ulex layer to host-port stored in cdat
		   (res      (send-receive uconn (conndat-hostport cdat) cmd params)))






	      (conndat-expires-set! cdat (+ (current-seconds)
					    (server:expiration-timeout)
					    -2)) ;; two second margin for network time misalignments etc.
	      res)))

;; db is at apath/.db/dbname, rid is an intermediary solution and will be removed
;; sometime in the future.
;;
;; Purpose - call the main.db server and request a server be started
;; for the given area path and dbname
;;

Modified ulex-simple/ulex.scm from [3fd48eb90f] to [0afbda11bc].

22
23
24
25
26
27
28
29


30
31
32
33
34
35
36
;;   See README in the distribution at https://www.kiatoa.com/fossils/ulex
;; NOTES:
;;   Why sql-de-lite and not say, dbi?  - performance mostly, then simplicity.
;;
;;======================================================================

(module ulex
    (


     
     ;; NOTE: looking for the handler proc - find the run-listener :)
     
     run-listener     ;; (run-listener handler-proc [port]) => uconn

     ;; NOTE: handler-proc params;
     ;;       (handler-proc rem-host-port qrykey cmd params)







<
>
>







22
23
24
25
26
27
28

29
30
31
32
33
34
35
36
37
;;   See README in the distribution at https://www.kiatoa.com/fossils/ulex
;; NOTES:
;;   Why sql-de-lite and not say, dbi?  - performance mostly, then simplicity.
;;
;;======================================================================

(module ulex

	*
	#;(
     
     ;; NOTE: looking for the handler proc - find the run-listener :)
     
     run-listener     ;; (run-listener handler-proc [port]) => uconn

     ;; NOTE: handler-proc params;
     ;;       (handler-proc rem-host-port qrykey cmd params)
48
49
50
51
52
53
54




55
56
57
58
59

60
61
62
63
64
65
66
67
68
69
70
71

72
73
74
75
76
77
78
     
     ;; needed to get the interface:port that was automatically found
     udat-port
     udat-host-port
     
     ;; for testing only
     ;; pp-uconn




     )

(import scheme
	chicken.base
	chicken.file

	chicken.time
	chicken.condition
	chicken.string
	chicken.sort
	chicken.pretty-print
	
	address-info
	mailbox
	matchable
	;; queues
	regex
	regex-case

	s11n
	srfi-1
	srfi-18
	srfi-4
	srfi-69
	system-information
	tcp6







>
>
>
>





>












>







49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
     
     ;; needed to get the interface:port that was automatically found
     udat-port
     udat-host-port
     
     ;; for testing only
     ;; pp-uconn
     
     ;; parameters
     work-method   ;; parameter; 'threads, 'mailbox, 'limited, 'direct
     return-method ;; parameter; 'mailbox, 'polling, 'direct
     )

(import scheme
	chicken.base
	chicken.file
	chicken.io
	chicken.time
	chicken.condition
	chicken.string
	chicken.sort
	chicken.pretty-print
	
	address-info
	mailbox
	matchable
	;; queues
	regex
	regex-case
	simple-exceptions
	s11n
	srfi-1
	srfi-18
	srfi-4
	srfi-69
	system-information
	tcp6
92
93
94
95
96
97
98
99
100
101

102
103
















104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120








121
122
123

124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161




162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198

























199
200
201
202
203


204
205
206


207
208
209


210



211
212

213




214
215
216


217



















218
219
220
221

222
223
224
225
226
227
228
229
230
231
232
233


















234
235
236

237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252

253
254


255










256
257

258
259






260
261
262
263

264
265

266
267
268
269
270


271

272
273
274
275
276

277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297

298
299
300

301
302
303
304











305
306


307
308
309
310
311
312
313
314
315

316
317






318
319
320


321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
  ;; work handling
  (work-queue (make-mailbox))
  (work-proc  #f)                ;; set by user
  (cnum       0)                 ;; cookie number
  (mboxes     (make-hash-table)) ;; for the replies
  (avail-cmboxes '())            ;; list of (<cookie> . <mbox>) for re-use
  ;; threads
  (numthreads 50)
  (cmd-thread #f)
  (work-queue-thread #f)

  ) 

















;; ;; struct for keeping track of others we are talking to
;; ;;
;; (defstruct pdat
;;   (host-port  #f)
;;   (conns      '()) ;; list of pcon structs, pop one off when calling the peer
;;   )
;; 
;; ;; struct for peer connections, keep track of expiration etc.
;; ;;
;; (defstruct pcon
;;   (inp #f)
;;   (oup #f)
;;   (exp (+ (current-seconds) 59)) ;; expires at this time, set to (+ (current-seconds) 59)
;;   (lifetime (+ (current-seconds) 600)) ;; throw away and create new after five minutes
;;   )

;;======================================================================








;; listener
;;======================================================================


;; is uconn a ulex connector (listener)
;;
(define (ulex-listener? uconn)
  (udat? uconn))

;; create a tcp listener and return a populated udat struct with
;; my port, address, hostname, pid etc.
;; return #f if fail to find a port to allocate.
;;
;;  if udata-in is #f create the record
;;  if there is already a serv-listener return the udata
;;
(define (setup-listener uconn #!optional (port 4242))
  (handle-exceptions
   exn
   (if (< port 65535)
       (setup-listener uconn (+ port 1))
       #f)
   (connect-listener uconn port)))

(define (connect-listener uconn port)
  ;; (tcp-listener-socket LISTENER)(socket-name so)
  ;; sockaddr-address, sockaddr-port, sockaddr->string
  (let* ((tlsn (tcp-listen port 1000 #f)) ;; (tcp-listen TCPPORT [BACKLOG [HOST]])
	 (addr (get-my-best-address))) ;; (hostinfo-addresses (host-information (current-hostname)))
    (udat-port-set!      uconn port)
    (udat-host-port-set! uconn (conc addr":"port))
    (udat-socket-set!    uconn tlsn)
    uconn))

;; run-listener does all the work of starting a listener in a thread
;; it then returns control
;;
(define (run-listener handler-proc #!optional (port-suggestion 4242))
  (let* ((uconn (make-udat)))
    (udat-work-proc-set! uconn handler-proc)
    (if (setup-listener uconn port-suggestion)
	(let* ((th1 (make-thread (lambda ()(ulex-cmd-loop uconn)) "Ulex command loop"))




	       #;(th2 (make-thread (lambda ()(process-work-queue uconn)) "Ulex work queue processor")))
	  (tcp-buffer-size 2048)
	  ;; (max-connections 2048) 
	  (thread-start! th1)
	  #;(thread-start! th2)
	  (udat-cmd-thread-set! uconn th1)
	  #;(udat-work-queue-thread-set! uconn th2)
	  (print "cmd loop and process workers started")
	  uconn)
	(assert #f "ERROR: run-listener called without proper setup."))))

(define (wait-and-close uconn)
  (thread-join! (udat-cmd-thread uconn))
  (tcp-close (udat-socket uconn)))

;;======================================================================
;; peers and connections
;;======================================================================

(define *send-mutex* (make-mutex))

;; send structured data to recipient
;;
;;  NOTE: qrykey is what was called the "cookie" previously
;;
;;     retval tells send to expect and wait for return data (one line) and return it or time out
;;       this is for ping where we don't want to necessarily have set up our own server yet.
;;
;; NOTE: see below for beginnings of code to allow re-use of tcp connections
;;        - I believe (without substantial evidence) that re-using connections will
;;          be beneficial ...
;;
(define (send udata host-port qrykey cmd params)
  (mutex-lock! *send-mutex*)
  (let* ((my-host-port (udat-host-port udata))          ;; remote will return to this
	 (isme         #f #;(equal? host-port my-host-port)) ;; calling myself?
	 ;; dat is a self-contained work block that can be sent or handled locally

























	 (dat          (list my-host-port qrykey cmd params)))
    (if isme
	(ulex-handler udata dat) ;; no transmission needed
	(handle-exceptions ;; TODO - MAKE THIS EXCEPTION CMD SPECIFIC?
	    exn


	    #f
	  (let-values (((inp oup)(tcp-connect host-port)))
	    (let ((res (if (and inp oup)


			   (begin
			     (serialize dat oup)
			     (deserialize inp)) ;; yes, we always want an ack


			   (begin



			     (print "ERROR: send called but no receiver has been setup. Please call setup first!")
			     #f))))

	      (close-input-port inp)




	      (close-output-port oup)
	      (mutex-unlock! *send-mutex*)
	      res)))))) ;; res will always be 'ack






















;; send a request to the given host-port and register a mailbox in udata
;; wait for the mailbox data and return it
;;
(define (send-receive uconn host-port cmd data)

  (cond
   ((member cmd '(ping goodbye)) ;; these are immediate
    (send uconn host-port 'ping cmd data))
   (else
    (let* ((cmbox     (get-cmbox uconn)) ;; would it be better to keep a stack of mboxes to reuse?
	   (qrykey    (car cmbox))
	   (mbox      (cdr cmbox))
	   (mbox-time (current-milliseconds))
	   (sres      (send uconn host-port qrykey cmd data))) ;; short res
      sres))))

;;======================================================================


















;; responder side
;;======================================================================


;; take a request, rdat, and if not immediate put it in the work queue
;;
;; Reserved cmds; ack ping goodbye response
;;
(define (ulex-handler uconn rdat)
  (assert (list? rdat) "FATAL: ulex-handler give rdat as not list")
  (match rdat ;;  (string-split controldat)
    ((rem-host-port qrykey cmd params)
     ;; (print "ulex-handler got: "rem-host-port" qrykey: "qrykey" cmd: "cmd" params: "params)
     (let ((mbox (hash-table-ref/default (udat-mboxes uconn) qrykey #f)))
       (case cmd
	 ;; ((ack )(print "Got ack! But why? Should NOT get here.") 'ack)
	 ((ping)
	  ;; (print "Got Ping!")
	  ;; (add-to-work-queue uconn rdat)
	 'ack)

	 (else
	  (do-work uconn rdat)))))


    (else










     (print "BAD DATA? controldat=" rdat)
     'ack) ;; send ack anyway?

    ))







;; given an already set up uconn start the cmd-loop
;;
(define (ulex-cmd-loop uconn)
  (let* ((serv-listener (udat-socket uconn)))

    (let loop ((state 'start))
      (let-values (((inp oup)(tcp-accept serv-listener)))

	(let* ((rdat  (deserialize inp)) ;; '(my-host-port qrykey cmd params)
	       (resp  (ulex-handler uconn rdat)))
	  (if resp (serialize resp oup))
	  (close-input-port inp)
	  (close-output-port oup))


	(loop state)))))

;;(define (ulex-cmd-loop uconn)
;;  (let* ((serv-listener (udat-socket uconn))
;;	 ;; (old-listener      (lambda ()
;;	 ;; 		      (let loop ((state 'start))
;;	 ;; 			(let-values (((inp oup)(tcp-accept serv-listener)))

;;	 ;; 			  (let* ((rdat  (deserialize inp)) ;; '(my-host-port qrykey cmd params)
;;	 ;; 				 (resp  (ulex-handler uconn rdat)))
;;	 ;; 			    (if resp (serialize resp oup))
;;	 ;; 			    (close-input-port inp)
;;	 ;; 			    (close-output-port oup))
;;	 ;; 			  (loop state)))))
;;	 (server       (make-tcp-server
;;			serv-listener
;;			(lambda ()
;;			  (let* ((rdat  (deserialize )) ;; '(my-host-port qrykey cmd params)
;;				 (resp  (ulex-handler uconn rdat)))
;;			    (if resp (serialize resp) resp))))))
;;    (server)))

;; add a proc to the cmd list, these are done symetrically (i.e. in all instances)
;; so that the proc can be dereferenced remotely
;;
(define (set-work-handler uconn proc)
  (udat-work-proc-set! uconn proc))

;;======================================================================

;; work queues - this is all happening on the listener side
;;======================================================================


;; rdat is (rem-host-port qrykey cmd params)
					     
(define (add-to-work-queue uconn rdat)
  #;(queue-add! (udat-work-queue uconn) rdat)











  (mailbox-send! (udat-work-queue uconn) rdat))



(define (do-work uconn rdat)
  (let* ((proc (udat-work-proc uconn))) ;; get it each time - conceivebly it could change
    ;; put this following into a do-work procedure
    (match rdat
      ((rem-host-port qrykey cmd params)
       (let* ((start-time (current-milliseconds))
	      (result (proc rem-host-port qrykey cmd params))
	      (end-time (current-milliseconds))
	      (run-time (- end-time start-time)))

	 result))
      (else






       (print "ERROR: rdat "rdat", did not match rem-host-port qrykey cmd params")
       #f))))
     


(define (process-work-queue uconn) 
  (let ((wqueue (udat-work-queue uconn))
	(proc   (udat-work-proc  uconn))
	(numthr (udat-numthreads uconn)))
    (let loop ((thnum    1)
	       (threads '()))
      (let ((thlst (cons (make-thread (lambda ()
					(let work-loop ()
					  (let ((rdat (mailbox-receive! wqueue 24000 'MBOX_TIMEOUT)))
					    (do-work uconn rdat))
					  (work-loop)))
				      (conc "work thread " thnum))
			 threads)))
	(if (< thnum numthr)
	    (loop (+ thnum 1)
		  thlst)
	    (begin
	      (print "ULEX: Starting "(length thlst)" worker threads.")
	      (map thread-start! thlst)
	      (print "ULEX: Threads started. Joining all.")
	      (map thread-join! thlst)))))))

;; below was to enable re-use of connections. This seems non-trivial so for
;; now lets open on each call
;;
;; ;; given host-port get or create peer struct
;; ;;
;; (define (udat-get-peer uconn host-port)
;;   (or (hash-table-ref/default (udat-peers uconn) host-port #f)
;;       ;; no peer, so create pdat and init it
;;       
;;       ;; NEED stack of connections, pop and use; inp, oup,
;;       ;; creation_time (remove and create new if over 24hrs old
;;       ;; 
;;       (let ((pdat (make-pdat host-port: host-port)))
;; 	(hash-table-set! (udat-peers uconn) host-port pdat)
;; 	pdat)))
;; 
;; ;; is pcon alive
;; 
;; ;; given host-port and pdat get a pcon
;; ;;
;; (define (pdat-get-pcon pdat host-port)
;;   (let loop ((conns (pdat-conns pdat)))
;;     (if (null? conns) ;; none? make and return - do NOT add - it will be pushed back on list later
;; 	(init-pcon (make-pcon))
;; 	(let* ((conn (pop conns)))
;; 	  
;; ;; given host-port get a pcon struct
;; ;;
;; (define (udat-get-pcon 
      
;;======================================================================
;; misc utils
;;======================================================================

(define (make-cookie uconn)
  (let ((newcnum (+ (udat-cnum uconn) 1)))
    (udat-cnum-set! uconn newcnum)
    (conc (udat-host-port uconn) ":"
	  newcnum)))

;; cookie/mboxes

;; we store each mbox with a cookie (<cookie> . <mbox>)
;;
(define (get-cmbox uconn)
  (if (null? (udat-avail-cmboxes uconn))
      (let ((cookie (make-cookie uconn))
	    (mbox   (make-mailbox)))
	(hash-table-set! (udat-mboxes uconn) cookie mbox)
	`(,cookie . ,mbox))
      (let ((cmbox (car (udat-avail-cmboxes uconn))))
	(udat-avail-cmboxes-set! uconn (cdr (udat-avail-cmboxes uconn)))
	cmbox)))

(define (put-cmbox uconn cmbox)
  (udat-avail-cmboxes-set! uconn (cons cmbox (udat-avail-cmboxes uconn))))

(define (pp-uconn uconn)
  (pp (udat->alist uconn)))

  
;;======================================================================
;; network utilities
;;======================================================================

;; NOTE: Look at address-info egg as alternative to some of this

(define (rate-ip ipaddr)
  (regex-case ipaddr
    ( "^127\\..*" _ 0 )
    ( "^(10\\.0|192\\.168)\\..*" _ 1 )
    ( else 2 ) ))

;; Change this to bias for addresses with a reasonable broadcast value?
;;
(define (ip-pref-less? a b)
  (> (rate-ip a) (rate-ip b)))

(define (get-my-best-address)
  (let ((all-my-addresses (get-all-ips)))
    (cond
     ((null? all-my-addresses)
      (get-host-name))                                          ;; no interfaces?
     ((eq? (length all-my-addresses) 1)
      (car all-my-addresses))                      ;; only one to choose from, just go with it
     (else
      (car (sort all-my-addresses ip-pref-less?))))))

(define (get-all-ips-sorted)
  (sort (get-all-ips) ip-pref-less?))

(define (get-all-ips)
  (map address-info-host
       (filter (lambda (x)
		 (equal? (address-info-type x) "tcp"))
	       (address-infos (get-host-name)))))

)







|


>


>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
|
|
|
|
|
|
<
<
<
<
<
<
<
<
|
|
|
>
>
>
>
>
>
>
>
|
<
|
>
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
>
>
>
>
|
|
<
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
<
|
|
|
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
|
|
|
|
|
>
>
|
<
<
>
>
|
<
<
>
>
|
>
>
>
|
|
>
|
>
>
>
>
|
<
|
>
>
|
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
|
|
|
|
>
|
|
|
<
<
<
|
<
<
<
|
|
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
|
<
|
>
|
|
|
|
|
|
|
|
|
<
|
|
|
|
|
|
>
|
|
>
>
|
>
>
>
>
>
>
>
>
>
>
|
|
>
|
|
>
>
>
>
>
>
|
|
|
|
>
|
|
>
|
|
|
|
|
>
>
|
>
|
<
<
|
<
>
|
<
<
|
<
|
<
<
<
|
<
|
<
|
|
|
|
|
|
|
<
>
|
<
|
>
|
|
|
|
>
>
>
>
>
>
>
>
>
>
>
|
|
>
>
|
|
|
|
|
|
|
|
|
>
|
|
>
>
>
>
>
>
|
<
|
>
>
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|

99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133








134
135
136
137
138
139
140
141
142
143
144
145

146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191

192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221

222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257


258
259
260


261
262
263
264
265
266
267
268
269
270
271
272
273
274
275

276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306



307



308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328

329
330
331
332
333
334
335
336
337
338
339

340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390


391

392
393


394

395



396

397

398
399
400
401
402
403
404

405
406

407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446

447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
  ;; work handling
  (work-queue (make-mailbox))
  (work-proc  #f)                ;; set by user
  (cnum       0)                 ;; cookie number
  (mboxes     (make-hash-table)) ;; for the replies
  (avail-cmboxes '())            ;; list of (<cookie> . <mbox>) for re-use
  ;; threads
  (numthreads 10)
  (cmd-thread #f)
  (work-queue-thread #f)
  (num-threads-running 0)
  ) 

;; == << ;; Parameters
;; == << 
;; == << ;; work-method:
;; == << (define work-method (make-parameter 'mailbox))
;; == << ;;    mailbox - all rdat goes through mailbox
;; == << ;;    threads - all rdat immediately executed in new thread
;; == << ;;    direct  - no queuing
;; == << ;;
;; == << 
;; == << ;; return-method, return the result to waiting send-receive:
;; == << (define return-method (make-parameter 'mailbox))
;; == << ;;    mailbox - create a mailbox and use it for passing returning results to send-receive
;; == << ;;    polling - put the result in a hash table keyed by qrykey and send-receive can poll it for result
;; == << ;;    direct  - no queuing, result is passed back in single tcp connection
;; == << ;;
;; == << 
;; == << ;; ;; struct for keeping track of others we are talking to
;; == << ;; ;;
;; == << ;; (defstruct pdat
;; == << ;;   (host-port  #f)
;; == << ;;   (conns      '()) ;; list of pcon structs, pop one off when calling the peer
;; == << ;;   )








;; == << ;; 
;; == << ;; ;; struct for peer connections, keep track of expiration etc.
;; == << ;; ;;
;; == << ;; (defstruct pcon
;; == << ;;   (inp #f)
;; == << ;;   (oup #f)
;; == << ;;   (exp (+ (current-seconds) 59)) ;; expires at this time, set to (+ (current-seconds) 59)
;; == << ;;   (lifetime (+ (current-seconds) 600)) ;; throw away and create new after five minutes
;; == << ;;   )
;; == << 
;; == << ;;======================================================================
;; == << ;; listener

;; == << ;;======================================================================
;; == << 
;; == << ;; is uconn a ulex connector (listener)
;; == << ;;
;; == << (define (ulex-listener? uconn)
;; == <<   (udat? uconn))
;; == << 
;; == << ;; create a tcp listener and return a populated udat struct with
;; == << ;; my port, address, hostname, pid etc.
;; == << ;; return #f if fail to find a port to allocate.
;; == << ;;
;; == << ;;  if udata-in is #f create the record
;; == << ;;  if there is already a serv-listener return the udata
;; == << ;;
;; == << (define (setup-listener uconn #!optional (port 4242))
;; == <<   (handle-exceptions
;; == <<    exn
;; == <<    (if (< port 65535)
;; == <<        (setup-listener uconn (+ port 1))
;; == <<        #f)
;; == <<    (connect-listener uconn port)))
;; == << 
;; == << (define (connect-listener uconn port)
;; == <<   ;; (tcp-listener-socket LISTENER)(socket-name so)
;; == <<   ;; sockaddr-address, sockaddr-port, sockaddr->string
;; == <<   (let* ((tlsn (tcp-listen port 1000 #f)) ;; (tcp-listen TCPPORT [BACKLOG [HOST]])
;; == << 	 (addr (get-my-best-address))) ;; (hostinfo-addresses (host-information (current-hostname)))
;; == <<     (udat-port-set!      uconn port)
;; == <<     (udat-host-port-set! uconn (conc addr":"port))
;; == <<     (udat-socket-set!    uconn tlsn)
;; == <<     uconn))
;; == << 
;; == << ;; run-listener does all the work of starting a listener in a thread
;; == << ;; it then returns control
;; == << ;;
;; == << (define (run-listener handler-proc #!optional (port-suggestion 4242))
;; == <<   (let* ((uconn (make-udat)))
;; == <<     (udat-work-proc-set! uconn handler-proc)
;; == <<     (if (setup-listener uconn port-suggestion)
;; == << 	(let* ((th1 (make-thread (lambda ()(ulex-cmd-loop uconn)) "Ulex command loop"))
;; == << 	       (th2 (make-thread (lambda ()
;; == << 				   (case (work-method)
;; == << 				     ((mailbox limited)
;; == << 				      (process-work-queue uconn))))
;; == << 				 "Ulex work queue processor")))
;; == << 	  ;; (tcp-buffer-size 2048)

;; == << 	  (thread-start! th1)
;; == << 	  (thread-start! th2)
;; == << 	  (udat-cmd-thread-set! uconn th1)
;; == << 	  (udat-work-queue-thread-set! uconn th2)
;; == << 	  (print "cmd loop and process workers started, listening on "(udat-host-port uconn)".")
;; == << 	  uconn)
;; == << 	(assert #f "ERROR: run-listener called without proper setup."))))
;; == << 
;; == << (define (wait-and-close uconn)
;; == <<   (thread-join! (udat-cmd-thread uconn))
;; == <<   (tcp-close (udat-socket uconn)))
;; == << 
;; == << ;;======================================================================
;; == << ;; peers and connections
;; == << ;;======================================================================
;; == << 
;; == << (define *send-mutex* (make-mutex))
;; == << 
;; == << ;; send structured data to recipient
;; == << ;;
;; == << ;;  NOTE: qrykey is what was called the "cookie" previously
;; == << ;;
;; == << ;;     retval tells send to expect and wait for return data (one line) and return it or time out
;; == << ;;       this is for ping where we don't want to necessarily have set up our own server yet.
;; == << ;;
;; == << ;; NOTE: see below for beginnings of code to allow re-use of tcp connections
;; == << ;;        - I believe (without substantial evidence) that re-using connections will
;; == << ;;          be beneficial ...
;; == << ;;
;; == << (define (send udata host-port qrykey cmd params)

;; == <<   (let* ((my-host-port (udat-host-port udata))          ;; remote will return to this
;; == << 	 (isme         #f #;(equal? host-port my-host-port)) ;; calling myself?
;; == << 	 ;; dat is a self-contained work block that can be sent or handled locally
;; == << 	 (dat          (list my-host-port qrykey cmd params #;(cons (current-seconds)(current-milliseconds)))))
;; == <<     (cond
;; == <<      (isme (ulex-handler udata dat)) ;; no transmission needed
;; == <<      (else
;; == <<       (handle-exceptions ;; TODO - MAKE THIS EXCEPTION CMD SPECIFIC?
;; == << 	  exn
;; == << 	  (message exn)
;; == << 	(begin
;; == << 	  ;; (mutex-lock! *send-mutex*) ;; DOESN'T SEEM TO HELP
;; == << 	  (let-values (((inp oup)(tcp-connect host-port)))
;; == << 	    (let ((res (if (and inp oup)
;; == << 			   (begin
;; == << 			     (serialize dat oup)
;; == << 			     (close-output-port oup)
;; == << 			     (deserialize inp)
;; == << 			     )
;; == << 			   (begin
;; == << 			     (print "ERROR: send called but no receiver has been setup. Please call setup first!")
;; == << 			     #f))))
;; == << 	      (close-input-port inp)
;; == << 	      ;; (mutex-unlock! *send-mutex*) ;; DOESN'T SEEM TO HELP
;; == << 	      res)))))))) ;; res will always be 'ack unless return-method is direct
;; == << 
;; == << (define (send-via-polling uconn host-port cmd data)
;; == <<   (let* ((qrykey (make-cookie uconn))
;; == << 	 (sres   (send uconn host-port qrykey cmd data)))
;; == <<     (case sres
;; == <<       ((ack)
;; == <<        (let loop ((start-time (current-milliseconds)))
;; == << 	 (if (> (current-milliseconds)(+ start-time 10000)) ;; ten seconds timeout
;; == << 	     (begin
;; == << 	       (print "ULEX ERROR: timed out waiting for response from "host-port", "cmd" "data)
;; == << 	       #f)


;; == << 	     (let* ((result (hash-table-ref/default (udat-mboxes uconn) qrykey #f))) ;; NOTE: we are re-using mboxes hash
;; == << 	       (if result ;; result is '(status . result-data) or #f for nothing yet
;; == << 		   (begin


;; == << 		     (hash-table-delete! (udat-mboxes uconn) qrykey)
;; == << 		     (cdr result))
;; == << 		   (begin
;; == << 		     (thread-sleep! 0.01)
;; == << 		     (loop start-time)))))))
;; == <<       (else
;; == <<        (print "ULEX ERROR: Communication failed? sres="sres)
;; == <<        #f))))
;; == << 
;; == << (define (send-via-mailbox uconn host-port cmd data)
;; == <<   (let* ((cmbox     (get-cmbox uconn)) ;; would it be better to keep a stack of mboxes to reuse?
;; == << 	 (qrykey    (car cmbox))
;; == << 	 (mbox      (cdr cmbox))
;; == << 	 (mbox-time (current-milliseconds))
;; == << 	 (sres      (send uconn host-port qrykey cmd data))) ;; short res

;; == <<     (if (eq? sres 'ack)
;; == << 	(let* ((mbox-timeout-secs    120 #;(if (eq? 'primordial (thread-name (current-thread)))
;; == << 				     #f
;; == << 				     120)) ;; timeout)
;; == << 	       (mbox-timeout-result 'MBOX_TIMEOUT)
;; == << 	       (res                  (mailbox-receive! mbox mbox-timeout-secs mbox-timeout-result))
;; == << 	       (mbox-receive-time    (current-milliseconds)))
;; == << 	  ;; (put-cmbox uconn cmbox) ;; reuse mbox and cookie. is it worth it?
;; == << 	  (hash-table-delete! (udat-mboxes uconn) qrykey)
;; == << 	  (if (eq? res 'MBOX_TIMEOUT)
;; == << 	      (begin
;; == << 		(print "WARNING: mbox timed out for query "cmd", with data "data
;; == << 		       ", waiting for response from "host-port".")
;; == << 
;; == << 		;; here it might make sense to clean up connection records and force clean start?
;; == << 		;; NO. The progam using ulex needs to do the reset. Right thing here is exception
;; == << 		
;; == << 		#f)  ;; convert to raising exception?
;; == << 	      res))
;; == << 	(begin
;; == << 	  (print "ERROR: Communication failed? Got "sres)
;; == << 	  #f))))
;; == <<   
;; == << ;; send a request to the given host-port and register a mailbox in udata
;; == << ;; wait for the mailbox data and return it
;; == << ;;
;; == << (define (send-receive uconn host-port cmd data)
;; == <<   (let* ((start-time (current-milliseconds))
;; == << 	 (result     (cond
;; == << 		      ((member cmd '(ping goodbye)) ;; these are immediate
;; == << 		       (send uconn host-port 'ping cmd data))



;; == << 		      ((eq? (work-method) 'direct)



;; == << 		       ;; the result from send will be the actual result, not an 'ack
;; == << 		       (send uconn host-port 'direct cmd data))
;; == << 		      (else
;; == << 		       (case (return-method)
;; == << 			 ((polling)
;; == << 			  (send-via-polling uconn host-port cmd data))
;; == << 			 ((mailbox) 
;; == << 			  (send-via-mailbox uconn host-port cmd data))
;; == << 			 (else
;; == << 			  (print "ULEX ERROR: unrecognised return-method "(return-method)".")
;; == << 			  #f)))))
;; == << 	 (duration    (- (current-milliseconds) start-time)))
;; == <<     ;; this is ONLY for development and debugging. It will be removed once Ulex is stable.
;; == <<     (if (< 5000 duration)
;; == << 	(print "ULEX WARNING: round-trip took "(inexact->exact (round (/ duration 1000)))
;; == << 	       " seconds; "cmd", host-port="host-port", data="data))
;; == <<     result))
;; == <<     
;; == << 
;; == << ;;======================================================================
;; == << ;; responder side

;; == << ;;======================================================================
;; == << 
;; == << ;; take a request, rdat, and if not immediate put it in the work queue
;; == << ;;
;; == << ;; Reserved cmds; ack ping goodbye response
;; == << ;;
;; == << (define (ulex-handler uconn rdat)
;; == <<   (assert (list? rdat) "FATAL: ulex-handler give rdat as not list")
;; == <<   (match rdat ;;  (string-split controldat)
;; == <<     ((rem-host-port qrykey cmd params);; timedata)
;; == <<      ;; (print "ulex-handler got: "rem-host-port" qrykey: "qrykey" cmd: "cmd" params: "params)

;; == <<      (case cmd
;; == <<        ;; ((ack )(print "Got ack! But why? Should NOT get here.") 'ack)
;; == <<        ((ping)
;; == << 	;; (print "Got Ping!")
;; == << 	;; (add-to-work-queue uconn rdat)
;; == << 	'ack)
;; == <<        ((goodbye)
;; == << 	;; just clear out references to the caller. NOT COMPLETE
;; == << 	(add-to-work-queue uconn rdat)
;; == << 	'ack)
;; == <<        ((response) ;; this is a result from remote processing, send it as mail ...
;; == << 	(case (return-method)
;; == << 	  ((polling)
;; == << 	   (hash-table-set! (udat-mboxes uconn) qrykey (cons 'ok params))
;; == << 	   'ack)
;; == << 	  ((mailbox)
;; == << 	   (let ((mbox (hash-table-ref/default (udat-mboxes uconn) qrykey #f)))
;; == << 	     (if mbox
;; == << 		 (begin
;; == << 		   (mailbox-send! mbox params) ;; params here is our result
;; == << 		   'ack)
;; == << 		 (begin
;; == << 		   (print "ERROR: received result but no associated mbox for cookie "qrykey)
;; == << 		   'no-mbox-found))))
;; == << 	  (else (print "ULEX ERROR: unrecognised return-method "(return-method))
;; == << 		'bad-return-method)))
;; == <<        (else ;; generic request - hand it to the work queue
;; == << 	(add-to-work-queue uconn rdat)
;; == << 	'ack)))
;; == <<     (else
;; == <<      (print "ULEX ERROR: bad rdat "rdat)
;; == <<      'bad-rdat)))
;; == << 
;; == << ;; given an already set up uconn start the cmd-loop
;; == << ;;
;; == << (define (ulex-cmd-loop uconn)
;; == <<   (let* ((serv-listener (udat-socket uconn))
;; == << 	 (listener      (lambda ()
;; == << 			  (let loop ((state 'start))
;; == << 			    (let-values (((inp oup)(tcp-accept serv-listener)))
;; == << 			      ;; (mutex-lock! *send-mutex*) ;; DOESN'T SEEM TO HELP
;; == << 			      (let* ((rdat  (deserialize inp)) ;; '(my-host-port qrykey cmd params)
;; == << 				     (resp  (ulex-handler uconn rdat)))
;; == << 				(serialize resp oup)
;; == << 				(close-input-port inp)
;; == << 				(close-output-port oup)
;; == << 				;; (mutex-unlock! *send-mutex*) ;; DOESN'T SEEM TO HELP
;; == << 				)
;; == << 			      (loop state))))))
;; == <<     ;; start N of them
;; == <<     (let loop ((thnum   0)


;; == << 	       (threads '()))

;; == <<       (if (< thnum 100)
;; == << 	  (let* ((th (make-thread listener (conc "listener" thnum))))


;; == << 	    (thread-start! th)

;; == << 	    (loop (+ thnum 1)



;; == << 		  (cons th threads)))

;; == << 	  (map thread-join! threads)))))

;; == << 
;; == << ;; add a proc to the cmd list, these are done symetrically (i.e. in all instances)
;; == << ;; so that the proc can be dereferenced remotely
;; == << ;;
;; == << (define (set-work-handler uconn proc)
;; == <<   (udat-work-proc-set! uconn proc))
;; == << 

;; == << ;;======================================================================
;; == << ;; work queues - this is all happening on the listener side

;; == << ;;======================================================================
;; == << 
;; == << ;; rdat is (rem-host-port qrykey cmd params)
;; == << 					     
;; == << (define (add-to-work-queue uconn rdat)
;; == <<   #;(queue-add! (udat-work-queue uconn) rdat)
;; == <<   (case (work-method)
;; == <<     ((threads)
;; == <<      (thread-start! (make-thread (lambda ()
;; == << 				   (do-work uconn rdat))
;; == << 				 "worker thread")))
;; == <<     ((mailbox)
;; == <<      (mailbox-send! (udat-work-queue uconn) rdat))
;; == <<     ((direct)
;; == <<      (do-work uconn rdat))
;; == <<     (else
;; == <<      (print "ULEX ERROR: work-method "(work-method)" not recognised, using mailbox.")
;; == <<      (mailbox-send! (udat-work-queue uconn) rdat))))
;; == <<      
;; == << ;; move the logic to return the result somewhere else?
;; == << ;;
;; == << (define (do-work uconn rdat)
;; == <<   (let* ((proc (udat-work-proc uconn))) ;; get it each time - conceivebly it could change
;; == <<     ;; put this following into a do-work procedure
;; == <<     (match rdat
;; == <<       ((rem-host-port qrykey cmd params)
;; == <<        (let* ((start-time (current-milliseconds))
;; == << 	      (result (proc rem-host-port qrykey cmd params))
;; == << 	      (end-time (current-milliseconds))
;; == << 	      (run-time (- end-time start-time)))
;; == << 	 (case (work-method)
;; == << 	   ((direct) result)
;; == << 	   (else
;; == << 	    (print "ULEX: work "cmd", "params" done in "run-time" ms")
;; == << 	    ;; send 'response as cmd and result as params
;; == << 	    (send uconn rem-host-port qrykey 'response result) ;; could check for ack
;; == << 	    (print "ULEX: response sent back to "rem-host-port" in "(- (current-milliseconds) end-time))))))
;; == <<       (MBOX_TIMEOUT 'do-work-timeout)
;; == <<       (else
;; == <<        (print "ERROR: rdat "rdat", did not match rem-host-port qrykey cmd params")))))

;; == << 
;; == << ;; NEW APPROACH:
;; == << ;;   
;; == << (define (process-work-queue uconn) 
;; == <<   (let ((wqueue (udat-work-queue uconn))
;; == << 	(proc   (udat-work-proc  uconn))
;; == << 	(numthr (udat-numthreads uconn)))
;; == <<     (let loop ((thnum    1)
;; == << 	       (threads '()))
;; == <<       (let ((thlst (cons (make-thread (lambda ()
;; == << 					(let work-loop ()
;; == << 					  (let ((rdat (mailbox-receive! wqueue 24000 'MBOX_TIMEOUT)))
;; == << 					    (do-work uconn rdat))
;; == << 					  (work-loop)))
;; == << 				      (conc "work thread " thnum))
;; == << 			 threads)))
;; == << 	(if (< thnum numthr)
;; == << 	    (loop (+ thnum 1)
;; == << 		  thlst)
;; == << 	    (begin
;; == << 	      (print "ULEX: Starting "(length thlst)" worker threads.")
;; == << 	      (map thread-start! thlst)
;; == << 	      (print "ULEX: Threads started. Joining all.")
;; == << 	      (map thread-join! thlst)))))))
;; == << 
;; == << ;; below was to enable re-use of connections. This seems non-trivial so for
;; == << ;; now lets open on each call
;; == << ;;
;; == << ;; ;; given host-port get or create peer struct
;; == << ;; ;;
;; == << ;; (define (udat-get-peer uconn host-port)
;; == << ;;   (or (hash-table-ref/default (udat-peers uconn) host-port #f)
;; == << ;;       ;; no peer, so create pdat and init it
;; == << ;;       
;; == << ;;       ;; NEED stack of connections, pop and use; inp, oup,
;; == << ;;       ;; creation_time (remove and create new if over 24hrs old
;; == << ;;       ;; 
;; == << ;;       (let ((pdat (make-pdat host-port: host-port)))
;; == << ;; 	(hash-table-set! (udat-peers uconn) host-port pdat)
;; == << ;; 	pdat)))
;; == << ;; 
;; == << ;; ;; is pcon alive
;; == << ;; 
;; == << ;; ;; given host-port and pdat get a pcon
;; == << ;; ;;
;; == << ;; (define (pdat-get-pcon pdat host-port)
;; == << ;;   (let loop ((conns (pdat-conns pdat)))
;; == << ;;     (if (null? conns) ;; none? make and return - do NOT add - it will be pushed back on list later
;; == << ;; 	(init-pcon (make-pcon))
;; == << ;; 	(let* ((conn (pop conns)))
;; == << ;; 	  
;; == << ;; ;; given host-port get a pcon struct
;; == << ;; ;;
;; == << ;; (define (udat-get-pcon 
;; == <<       
;; == << ;;======================================================================
;; == << ;; misc utils
;; == << ;;======================================================================
;; == << 
;; == << (define (make-cookie uconn)
;; == <<   (let ((newcnum (+ (udat-cnum uconn) 1)))
;; == <<     (udat-cnum-set! uconn newcnum)
;; == <<     (conc (udat-host-port uconn) ":"
;; == << 	  newcnum)))
;; == << 
;; == << ;; cookie/mboxes
;; == << 
;; == << ;; we store each mbox with a cookie (<cookie> . <mbox>)
;; == << ;;
;; == << (define (get-cmbox uconn)
;; == <<   (if (null? (udat-avail-cmboxes uconn))
;; == <<       (let ((cookie (make-cookie uconn))
;; == << 	    (mbox   (make-mailbox)))
;; == << 	(hash-table-set! (udat-mboxes uconn) cookie mbox)
;; == << 	`(,cookie . ,mbox))
;; == <<       (let ((cmbox (car (udat-avail-cmboxes uconn))))
;; == << 	(udat-avail-cmboxes-set! uconn (cdr (udat-avail-cmboxes uconn)))
;; == << 	cmbox)))
;; == << 
;; == << (define (put-cmbox uconn cmbox)
;; == <<   (udat-avail-cmboxes-set! uconn (cons cmbox (udat-avail-cmboxes uconn))))
;; == << 
;; == << (define (pp-uconn uconn)
;; == <<   (pp (udat->alist uconn)))
;; == << 
;; == <<   
;; == << ;;======================================================================
;; == << ;; network utilities
;; == << ;;======================================================================
;; == << 
;; == << ;; NOTE: Look at address-info egg as alternative to some of this
;; == << 
;; == << (define (rate-ip ipaddr)
;; == <<   (regex-case ipaddr
;; == <<     ( "^127\\..*" _ 0 )
;; == <<     ( "^(10\\.0|192\\.168)\\..*" _ 1 )
;; == <<     ( else 2 ) ))
;; == << 
;; == << ;; Change this to bias for addresses with a reasonable broadcast value?
;; == << ;;
;; == << (define (ip-pref-less? a b)
;; == <<   (> (rate-ip a) (rate-ip b)))
;; == << 
;; == << (define (get-my-best-address)
;; == <<   (let ((all-my-addresses (get-all-ips)))
;; == <<     (cond
;; == <<      ((null? all-my-addresses)
;; == <<       (get-host-name))                                          ;; no interfaces?
;; == <<      ((eq? (length all-my-addresses) 1)
;; == <<       (car all-my-addresses))                      ;; only one to choose from, just go with it
;; == <<      (else
;; == <<       (car (sort all-my-addresses ip-pref-less?))))))
;; == << 
;; == << (define (get-all-ips-sorted)
;; == <<   (sort (get-all-ips) ip-pref-less?))
;; == << 
;; == << (define (get-all-ips)
;; == <<   (map address-info-host
;; == <<        (filter (lambda (x)
;; == << 		 (equal? (address-info-type x) "tcp"))
;; == << 	       (address-infos (get-host-name)))))
;; == << 
)