« back to all changes in this revision
Viewing changes to src/listen.cc
- browse files at revision 1.2.1
- compare with another revision
- download diff
- download tarball
- view history from revision 1.2.1
- Committer: Package Import Robot
- Author(s): Sébastien Villemot
- Date: 2012-10-17 13:40:55 UTC
- mfrom: (1.1.6)
- mto: (5.1.2 experimental)
- mto: This revision was merged to the branch mainline in revision 13.
- Revision ID: package-import@ubuntu.com-20121017134055-e8lrxjd3qgcd3kmt
Tags: upstream-1.2.0
Import upstream version 1.2.0
added
removed
src/listen.cc
1
#define STATUS(x) do { if (debug) std::cout << x << std::endl << std::flush; } while (0)
2
#define USE_DEFUN_INTERNAL 1
3
// Temporary hack (I hope). For some reason the interpreter
4
// is not finding the send/senderror commands, so I install
5
// them by hand..
6
7
#include <iomanip>
8
9
#include <cstdio>
10
#include <cctype>
11
#include <cstdlib>
12
#include <unistd.h>
13
//#include <stdint.h>
14
#include <cerrno>
15
// #include <string.h>
16
#include <sys/types.h>
17
#include <sys/socket.h>
18
#include <netinet/in.h>
19
#include <arpa/inet.h>
20
#include <sys/wait.h>
21
#include <signal.h>
22
23
#include <octave/oct.h>
24
#include <octave/parse.h>
25
#include <octave/variables.h>
26
#if 0
27
#include <octave/unwind-prot.h>
28
#endif
29
#include <octave/oct-syscalls.h>
30
#include <octave/oct-time.h>
31
#include <octave/lo-mappers.h>
32
#include <octave/symtab.h>
33
34
static bool debug = false;
35
static char* context = NULL;
36
37
static double timestamp = 0.0;
38
inline void tic(void) { timestamp = octave_time().double_value(); }
39
inline double toc(void) {return ceil(-1e6*(timestamp-octave_time().double_value()));}
40
41
// XXX FIXME XXX --- surely this is part of the standard library?
42
void
43
lowercase (std::string& s)
44
{
45
for (std::string::iterator i=s.begin(); i != s.end(); i++) *i = tolower(*i);
46
}
47
48
#if 0
49
octave_value
50
get_builtin_value (const std::string& nm)
51
{
52
octave_value retval;
53
54
symbol_record *sr = fbi_sym_tab->lookup (nm);
55
56
if (sr)
57
{
58
octave_value sr_def = sr->def ();
59
60
if (sr_def.is_undefined ())
61
error ("get_builtin_value: undefined symbol `%s'", nm.c_str ());
62
else
63
retval = sr_def;
64
}
65
else
66
error ("get_builtin_value: unknown symbol `$s'", nm.c_str ());
67
68
return retval;
69
}
70
#endif
71
72
// XXX FIXME XXX autoconf stuff
73
#if 0 && defined(_sgi)
74
typedef int socklen_t;
75
#endif
76
77
static void
78
sigchld_handler(int /* sig */)
79
{
80
int status;
81
/* Reap all childrens */
82
STATUS("reaping all children");
83
while (waitpid(-1, &status, WNOHANG) > 0)
84
;
85
STATUS("done reaping children");
86
}
87
88
/* Posix signal handling, based on the example from the
89
* Unix Programming FAQ
90
* Copyright (C) 2000 Andrew Gierth
91
*/
92
static void sigchld_setup(void)
93
{
94
struct sigaction act;
95
96
/* Assign sig_chld as our SIGCHLD handler */
97
act.sa_handler = sigchld_handler;
98
99
/* We don't want to block any other signals in this example */
100
sigemptyset(&act.sa_mask);
101
102
/*
103
* We're only interested in children that have terminated, not ones
104
* which have been stopped (eg user pressing control-Z at terminal)
105
*/
106
act.sa_flags = SA_NOCLDSTOP;
107
108
/*
109
* Make these values effective. If we were writing a real
110
* application, we would probably save the old value instead of
111
* passing NULL.
112
*/
113
if (sigaction(SIGCHLD, &act, NULL) < 0)
114
error("listen could not set SIGCHLD");
115
}
116
117
118
#if defined(__CYGWIN__)
119
120
// Don't daemonize on cygwin just yet.
121
inline void daemonize(void) {}
122
123
#else
124
125
static RETSIGTYPE
126
sigterm_handler(int /* sig */)
127
{
128
exit(0);
129
}
130
131
132
static void
133
daemonize(void)
134
{
135
if (fork()) exit(0);
136
std::cout << "Octave pid: " << octave_syscalls::getpid() << std::endl;
137
signal(SIGTERM,sigterm_handler);
138
signal(SIGQUIT,sigterm_handler);
139
140
freopen("/dev/null", "r", stdin);
141
freopen("/dev/null", "w", stdout);
142
freopen("/dev/null", "w", stderr);
143
}
144
145
#endif
146
147
148
149
static octave_value get_octave_value(char *name)
150
{
151
octave_value def;
152
153
// Copy variable from octave
154
#ifdef HAVE_OCTAVE_30
155
symbol_record *sr = top_level_sym_tab->lookup (name);
156
if (sr) def = sr->def();
157
#else
158
def = symbol_table::varref (std::string (name), symbol_table::top_scope ());
159
#endif
160
161
return def;
162
}
163
164
165
static void channel_error (const int channel, const char *str)
166
{
167
STATUS("sending error !!!e (" << strlen(str) << ") " << str);
168
169
uint32_t len = strlen(str);
170
write(channel,"!!!e",4);
171
uint32_t t = htonl(len); write(channel,&t,4);
172
write(channel,str,len);
173
}
174
175
static bool reads (const int channel, void * buf, int n)
176
{
177
// STATUS("entering reads loop with size " << n); tic();
178
while (1) {
179
int chunk = read(channel, buf, n);
180
if (chunk == 0) STATUS("read socket returned 0");
181
if (chunk < 0) STATUS("read socket: " << strerror(errno));
182
if (chunk <= 0) return false;
183
n -= chunk;
184
// if (n == 0) STATUS("done reads loop after " << toc() << "us");
185
if (n == 0) return true;
186
// STATUS("reading remaining " << n << " characters");
187
buf = (void *)((char *)buf + chunk);
188
}
189
}
190
191
static bool writes (const int channel, const void * buf, int n)
192
{
193
// STATUS("entering writes loop");
194
while (1) {
195
int chunk = write(channel, buf, n);
196
if (chunk == 0) STATUS("write socket returned 0");
197
if (chunk < 0) STATUS("write socket: " << strerror(errno));
198
if (chunk <= 0) return false;
199
n -= chunk;
200
// if (n == 0) STATUS("done writes loop");
201
if (n == 0) return true;
202
buf = (void *)((char *)buf + chunk);
203
}
204
}
205
206
static void
207
process_commands(int channel)
208
{
209
// XXX FIXME XXX check read/write return values
210
assert(sizeof(uint32_t) == 4);
211
char command[5];
212
char def_context[16536];
213
bool ok;
214
STATUS("waiting for command");
215
216
// XXX FIXME XXX do we need to specify the context size?
217
// int bufsize=sizeof(def_context);
218
// socklen_t ol;
219
// ol=sizeof(bufsize);
220
// setsockopt(channel,SOL_SOCKET,SO_SNDBUF,&bufsize,ol);
221
// setsockopt(channel,SOL_SOCKET,SO_RCVBUF,&bufsize,ol);
222
223
// XXX FIXME XXX prepare to capture long jumps, because if
224
// we dont, then errors in octave might escape to the prompt
225
226
command[4] = '\0';
227
if (debug) tic();
228
while (reads(channel, &command, 4)) {
229
// XXX FIXME XXX do whatever is require to check if function files
230
// have changed; do we really want to do this for _every_ command?
231
// Maybe we need a 'reload' command.
232
STATUS("received command " << command << " after " << toc() << "us");
233
234
// Check for magic command code
235
if (command[0] != '!' || command[1] != '!' || command[2] != '!') {
236
STATUS("communication error: closing connection");
237
break;
238
}
239
240
// Get command length
241
if (debug) tic(); // time the read
242
uint32_t len;
243
if (!reads(channel, &len, 4)) break;
244
len = ntohl(len);
245
// STATUS("read 4 byte command length in " << toc() << "us");
246
247
// Read the command context, allocating a new one if the default
248
// is too small.
249
if (len > (signed)sizeof(def_context)-1) {
250
// XXX FIXME XXX use octave allocators
251
// XXX FIXME XXX unwind_protect
252
context= new char[len+1];
253
if (context== NULL) {
254
// Requested command is too large --- skip to the next command
255
// XXX FIXME XXX maybe we want to kill the connection instead?
256
channel_error(channel,"out of memory");
257
ok = true;
258
STATUS("skip big command loop");
259
while (ok && len > (signed)sizeof(def_context)) {
260
ok = reads(channel, def_context, sizeof(def_context));
261
len -= sizeof(def_context);
262
}
263
STATUS("done skip big command loop");
264
if (!ok) break;
265
ok = reads(channel, def_context, sizeof(def_context));
266
if (!ok) break;
267
continue;
268
}
269
} else {
270
context = def_context;
271
}
272
// if (debug) tic();
273
ok = reads(channel, context, len);
274
context[len] = '\0';
275
STATUS("read " << len << " byte command in " << toc() << "us");
276
277
// Process the command
278
if (ok) switch (command[3]) {
279
case 'm': // send the named matrix
280
{
281
// XXX FIXME XXX this can be removed: app can do send(name,value)
282
STATUS("sending " << context);
283
uint32_t t;
284
285
// read the matrix contents
286
octave_value def = get_octave_value(context);
287
if(!def.is_defined() || !def.is_real_matrix())
288
channel_error(channel,"not a matrix");
289
Matrix m = def.matrix_value();
290
291
// write the matrix transfer header
292
ok = writes(channel,"!!!m",4); // matrix message
293
t = htonl(12 + sizeof(double)*m.rows()*m.columns());
294
if (ok) ok = writes(channel,&t,4); // length of message
295
t = htonl(m.rows());
296
if (ok) ok = writes(channel,&t,4); // rows
297
t = htonl(m.columns());
298
if (ok) ok = writes(channel,&t,4); // columns
299
t = htonl(len);
300
if (ok) ok = writes(channel, &t, 4); // name length
301
if (ok) ok = writes(channel,context,len); // name
302
303
// write the matrix contents
304
const double *v = m.data(); // data
305
if (ok) ok = writes(channel,v,sizeof(double)*m.rows()*m.columns());
306
if (ok)
307
STATUS("sent " << m.rows()*m.columns());
308
else
309
STATUS("failed " << m.rows()*m.columns());
310
}
311
break;
312
313
case 'x': // silently execute the command
314
{
315
if (debug)
316
{
317
if (len > 500)
318
{
319
// XXX FIXME XXX can we limit the maximum output width for a
320
// string? The setprecision() io manipulator doesn't do it.
321
// In the meantime, a hack ...
322
char t = context[400]; context[400] = '\0';
323
STATUS("evaluating (" << len << ") "
324
<< context << std::endl
325
<< "..." << std::endl
326
<< context+len-100);
327
context[400] = t;
328
}
329
else
330
{
331
STATUS("evaluating (" << len << ") " << context);
332
}
333
}
334
335
if (debug) tic();
336
#if 1
337
octave_value_list evalargs;
338
evalargs(1) = "senderror(lasterr);";
339
evalargs(0) = context;
340
octave_value_list fret = feval("eval",evalargs,0);
341
#else
342
evalargs(0) = octave_value(0.);
343
#endif
344
STATUS("done command");
345
}
346
STATUS("free evalargs");
347
break;
348
349
case 'c': // execute the command and capture stdin/stdout
350
STATUS("capture command not yet implemented");
351
break;
352
353
default:
354
STATUS("ignoring command " << command);
355
break;
356
}
357
358
if (context != def_context) delete[] context;
359
STATUS("done " << command);
360
if (!ok) break;
361
if (debug) tic();
362
}
363
}
364
365
366
int channel = -1;
367
368
#if USE_DEFUN_INTERNAL
369
DEFUN_INTERNAL(senderror,args,,false,"\
370
Send the given error message across the socket. The error context\n\
371
is taken to be the last command received from the socket.")
372
#else
373
DEFUN_DLD(senderror,args,,"\
374
Send the given error message across the socket. The error context\n\
375
is taken to be the last command received from the socket.")
376
#endif
377
{
378
std::string str;
379
const int nargin = args.length();
380
if (nargin != 1) str="senderror not called with error";
381
else str = args(0).string_value();
382
383
// provide a context for the error (but not too much!)
384
str += "when evaluating:\n";
385
if (strlen(context) > 100)
386
{
387
char t=context[100];
388
context[100] = '\0';
389
str+=context;
390
context[100]=t;
391
}
392
else
393
str += context;
394
395
STATUS("error is " << str);
396
channel_error(channel,str.c_str());
397
return octave_value_list();
398
}
399
400
#if USE_DEFUN_INTERNAL
401
DEFUN_INTERNAL(send,args,,false,"\
402
send(str)\n\
403
Send a command on the current connection\n\
404
send(name,value)\n\
405
Send a binary value with the given name on the current connection\n\
406
")
407
#else
408
DEFUN_DLD(send,args,,"\
409
send(str)\n\
410
Send a command on the current connection\n\
411
send(name,value)\n\
412
Send a binary value with the given name on the current connection\n\
413
")
414
#endif
415
{
416
bool ok;
417
uint32_t t;
418
octave_value_list ret;
419
int nargin = args.length();
420
if (nargin < 1 || nargin > 2)
421
{
422
print_usage ();
423
return ret;
424
}
425
426
if (channel < 0) {
427
error("Not presently listening on a port");
428
return ret;
429
}
430
431
std::string cmd(args(0).string_value());
432
if (error_state) return ret;
433
434
// XXX FIXME XXX perhaps process the panalopy of types?
435
if (nargin > 1) {
436
STATUS("sending !!!x(" << cmd.length() << ") " << cmd.c_str());
437
438
octave_value def = args(1);
439
if (args(1).is_string()) {
440
// Grab the string value from args(1).
441
// Can't use args(1).string_value() because that trims trailing \0
442
charMatrix m(args(1).char_matrix_value());
443
std::string s(m.row_as_string(0,false,true));
444
ok = writes(channel,"!!!s",4); // string message
445
t = htonl(8 + cmd.length() + s.length());
446
if (ok) ok = writes(channel,&t,4); // length of message
447
t = htonl(s.length());
448
if (ok) ok = writes(channel, &t, 4); // string length
449
t = htonl(cmd.length());
450
if (ok) ok = writes(channel, &t, 4); // name length
451
if (cmd.length() && ok)
452
ok = writes(channel, cmd.c_str(), cmd.length()); // name
453
if (s.length() && ok)
454
ok = writes(channel, s.c_str(), s.length()); // string
455
STATUS("sent string(" << s.length() << ")");
456
} else if (args(1).is_real_type()) {
457
Matrix m(args(1).matrix_value());
458
459
// write the matrix transfer header
460
ok = writes(channel,"!!!m",4); // matrix message
461
t = htonl(12 + cmd.length() + sizeof(double)*m.rows()*m.columns());
462
if (ok) ok = writes(channel,&t,4); // length of message
463
t = htonl(m.rows());
464
if (ok) ok = writes(channel,&t,4); // rows
465
t = htonl(m.columns());
466
if (ok) ok = writes(channel,&t,4); // columns
467
t = htonl(cmd.length());
468
if (ok) ok = writes(channel, &t, 4); // name length
469
if (ok) ok = writes(channel, cmd.c_str(), cmd.length()); // name
470
471
// write the matrix contents
472
const double *v = m.data(); // data
473
if (m.rows()*m.columns() && ok)
474
ok = writes(channel,v,sizeof(double)*m.rows()*m.columns());
475
STATUS("sent matrix(" << m.rows() << "x" << m.columns() << ")");
476
} else {
477
ok = false;
478
error("send expected name and matrix or string value");
479
}
480
if (!ok) error("send could not write to channel");
481
} else {
482
// STATUS("start writing at "<<toc()<<"us");
483
ok = writes(channel, "!!!x", 4);
484
t = htonl(cmd.length()); writes(channel, &t, 4);
485
if (ok) ok = writes(channel, cmd.c_str(), cmd.length());
486
if (!ok) error("send could not write to channel");
487
// STATUS("stop writing at "<<toc()<<"us");
488
}
489
490
return ret;
491
}
492
493
extern "C" int listencanfork(void);
494
extern "C" int StringCaseMatch(const char* s, const char* p, int nocase);
495
496
bool ishostglob(const std::string& s)
497
{
498
for (unsigned int i=0; i < s.length(); i++) {
499
if (! ( isdigit(s[i]) || s[i]=='*' || s[i]=='-'
500
|| s[i]=='.' || s[i]=='[' || s[i]==']')) return false;
501
}
502
return true;
503
}
504
505
bool anyhostglob(const string_vector& hostlist, const char* host)
506
{
507
for (int j=0; j < hostlist.length(); j++) {
508
if (StringCaseMatch(host, hostlist[j].c_str(), 0)) return true;
509
}
510
return false;
511
}
512
513
// Known bug: functions which pass or return structures use a
514
// different ABI for gcc and native compilers on some architectures.
515
// Whether this is a bug depends on the structure length. SGI's 64-bit
516
// architecture makes this a problem for inet_ntoa.
517
#if defined(__GNUC__) && defined(_sgi)
518
#define BROKEN_INET_NTOA
519
#endif
520
521
#ifdef BROKEN_INET_NTOA
522
523
/*************************************************
524
* Replacement for broken inet_ntoa() *
525
*************************************************/
526
527
528
/* On IRIX systems, gcc uses a different structure passing convention to the
529
native libraries. This causes inet_ntoa() to always yield 0.0.0.0 or
530
255.255.255.255. To get round this, we provide a private version of the
531
function here. It is used only if USE_INET_NTOA_FIX is set, which should
532
happen
533
only when gcc is in use on an IRIX system. Code send to me by J.T. Breitner,
534
with these comments:
535
536
537
code by Stuart Levy
538
as seen in comp.sys.sgi.admin
539
540
541
Arguments: sa an in_addr structure
542
Returns: pointer to static text string
543
*/
544
545
546
char *
547
inet_ntoa(struct in_addr sa)
548
{
549
static char addr[20];
550
sprintf(addr, "%d.%d.%d.%d",
551
(US &sa.s_addr)[0],
552
(US &sa.s_addr)[1],
553
(US &sa.s_addr)[2],
554
(US &sa.s_addr)[3]);
555
return addr;
556
}
557
558
#endif /* BROKEN_INET_NTOA */
559
560
561
DEFUN_DLD(listen,args,,"\
562
listen(port,host,host,...)\n\
563
Listen for connections on the given port. Normally only accepts\n\
564
connections from localhost (127.0.0.1), but you can specify any\n\
565
dot-separated host name globs. E.g., '128.2.20.*' or '128.2.2[012].*'\n\
566
Use '?' for '[0123456789]'. Use '*.*.*.*' for any host.\n\
567
listen(...,'debug'|'nodebug')\n\
568
If debug, echo all commands sent across the connection. If nodebug,\n\
569
detach the process and don't echo anything. You will need to use\n\
570
kill directly to end the process. Nodebug is the default.\n\
571
listen(...,'fork'|'nofork')\n\
572
If fork, start new server for each connection. If nofork, only allow\n\
573
one connection at a time. Fork is the default (depending on system).\n\
574
listen(...,'loopback')\n\
575
Use loopback address 127.0.0.1 rather than 0.0.0.0.\n\
576
")
577
{
578
#if USE_DEFUN_INTERNAL
579
install_builtin_function (Fsend, "send", "builtin send doc", false);
580
install_builtin_function (Fsenderror,
581
"senderror", "builtin senderror doc", false);
582
#endif
583
584
#if defined(__CYGWIN__)
585
bool canfork = listencanfork();
586
#else
587
bool canfork = true;
588
#endif
589
590
octave_value_list ret;
591
int nargin = args.length();
592
if (nargin < 1)
593
{
594
print_usage ();
595
return ret;
596
}
597
598
int port = args(0).int_value();
599
if (error_state) return ret;
600
601
debug = false;
602
uint32_t inaddr = INADDR_ANY;
603
604
string_vector hostlist;
605
hostlist.append(std::string("127.0.0.1"));
606
for (int k = 1; k < nargin; k++) {
607
std::string lastarg(args(k).string_value());
608
if (error_state) return ret;
609
lowercase(lastarg);
610
if (lastarg == "debug") {
611
debug = true;
612
} else if (lastarg == "nodebug") {
613
debug = false;
614
} else if (lastarg == "fork") {
615
canfork = true;
616
} else if (lastarg == "nofork") {
617
canfork = false;
618
} else if (lastarg == "loopback") {
619
inaddr = INADDR_LOOPBACK;
620
} else if (ishostglob(lastarg)) {
621
hostlist.append(lastarg);
622
} else {
623
print_usage ();
624
}
625
}
626
627
int sockfd; // listen on sockfd, new connection channel
628
struct sockaddr_in my_addr; // my address information
629
struct sockaddr_in their_addr; // connector's address information
630
socklen_t sin_size;
631
int yes=1;
632
633
if ((sockfd = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
634
perror("socket");
635
return ret;
636
}
637
638
if (setsockopt(sockfd,SOL_SOCKET,SO_REUSEADDR,&yes,sizeof(yes)) == -1) {
639
perror("setsockopt");
640
return ret;
641
}
642
643
my_addr.sin_family = AF_INET; // host byte order
644
my_addr.sin_port = htons(port); // short, network byte order
645
my_addr.sin_addr.s_addr = htonl(inaddr); // automatically fill with my IP
646
memset(&(my_addr.sin_zero), '\0', 8); // zero the rest of the struct
647
648
if (bind(sockfd, (struct sockaddr *)&my_addr, sizeof(struct sockaddr))
649
== -1) {
650
perror("bind");
651
close(sockfd);
652
return ret;
653
}
654
655
/* listen for connections (allowing one pending connection) */
656
if (listen(sockfd, canfork?1:0) == -1) {
657
perror("listen");
658
close(sockfd);
659
return ret;
660
}
661
662
#if 0
663
unwind_protect::begin_frame("Flisten");
664
unwind_protect_bool (buffer_error_messages);
665
buffer_error_messages = true;
666
#endif
667
668
sigchld_setup();
669
670
if (!debug && canfork) daemonize();
671
672
// XXX FIXME XXX want a 'sandbox' option which disables fopen, cd, pwd,
673
// system, popen ... Or maybe just an initial script to run for each
674
// connection, plus a separate command to disable specific functions.
675
STATUS("listening on port " << port);
676
while(1) { // main accept() loop
677
sin_size = sizeof(struct sockaddr_in);
678
STATUS("trying to accept");
679
if ((channel = accept(sockfd, (struct sockaddr *)&their_addr,
680
&sin_size)) == -1) {
681
// XXX FIXME XXX
682
// Linux is returning "Interrupted system call" when the
683
// child terminates. Until I figure out why, I can't use
684
// accept errors as a basis for breaking out of the listen
685
// loop, so instead print the octave PID so that I can kill
686
// it from another terminal.
687
STATUS("failed to accept" << std::endl
688
<< "Octave pid: " << octave_syscalls::getpid() );
689
perror("accept");
690
#if defined(_sgi)
691
break;
692
#else
693
continue;
694
#endif
695
}
696
STATUS("connected");
697
698
/* Simulate inet_ntoa */
699
const char *them = inet_ntoa(their_addr.sin_addr);
700
STATUS("server: got connection from " << them);
701
702
if (anyhostglob(hostlist,them)) {
703
if (canfork) {
704
int pid = fork();
705
706
if (pid == -1) {
707
perror("fork ");
708
break;
709
} else if (pid == 0) {
710
close(sockfd); // child doesn't need listener
711
signal(SIGCHLD,SIG_DFL); // child doesn't need SIGCHLD signal
712
process_commands(channel);
713
STATUS("child is exitting");
714
exit(0);
715
}
716
} else {
717
process_commands(channel);
718
STATUS("server: connection closed");
719
}
720
} else {
721
STATUS("server: connection refused.");
722
}
723
724
close(channel);
725
channel = -1;
726
}
727
728
STATUS("could not read commands; returning");
729
close(sockfd);
730
#if 0
731
unwind_protect::run_frame("Flisten");
732
#endif
733
return ret;
734
}
Loggerhead is a web-based interface for Breezy
Version: 2.0.1