Changeset: d7b9148d097e for MonetDB
URL: http://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=d7b9148d097e
Modified Files:
sql/backends/monet5/datacell/basket.mx
sql/backends/monet5/datacell/emitter.mx
Branch: default
Log Message:
Use UDP channels if you want them
Some admin cleanup to use UDP channels in the emitter when you intend so.
The first batches sent from the sensor were catched correctly by netcat.
diffs (184 lines):
diff --git a/sql/backends/monet5/datacell/basket.mx
b/sql/backends/monet5/datacell/basket.mx
--- a/sql/backends/monet5/datacell/basket.mx
+++ b/sql/backends/monet5/datacell/basket.mx
@@ -418,7 +418,7 @@
if (bskt == 0 )
throw(MAL,"basket.pass","Basket not found");
if ( baskets[bskt].colcount != pci->argc - 2)
- throw(MAL,"basket.pass","Non-matchin arguments");
+ throw(MAL,"basket.pass","Non-matching arguments");
/* copy the content of the temporary BATs into the basket */
mal_set_lock(baskets[bskt].lock, "lock basket");
@@ -426,7 +426,7 @@
ret= *(int*) getArgReference(stk,pci,j);
b = baskets[bskt].primary[i];
bn = BATdescriptor(ret);
- BATappend(b, bn, FALSE);
+ BATappend(b, bn, TRUE);
BBPreleaseref(ret);
}
mal_unset_lock(baskets[bskt].lock, "unlock basket");
diff --git a/sql/backends/monet5/datacell/emitter.mx
b/sql/backends/monet5/datacell/emitter.mx
--- a/sql/backends/monet5/datacell/emitter.mx
+++ b/sql/backends/monet5/datacell/emitter.mx
@@ -58,7 +58,7 @@
comment "Set the protocol of the emitter to udp/tcp/csv");
command drop(tb:str)
-address EMdrop
+address EMstop
comment "Drop a emitter";
command reset()
@@ -97,7 +97,7 @@
adapters_export str DCemitterNew(int *ret, str *tbl, str *host, int *port);
adapters_export str DCemitterPause(int *ret, str *nme);
adapters_export str DCemitterResume(int *ret, str *nme);
-adapters_export str EMdrop(int *ret, str *nme);
+adapters_export str EMstop(int *ret, str *nme);
adapters_export str EMreset(int *ret);
adapters_export str EMdump();
adapters_export str EMmode(int *ret, str *nme, str *arg);
@@ -128,8 +128,6 @@
static str protocolname[4] = {"<unknown>","TCP","UDP","CSV"};
-static int protocol= TCP;
-
#define PAUSEDEFAULT 1000
typedef struct EMITTER{
@@ -207,7 +205,9 @@
em->lck = 0;
em->error = NULL;
em->mode = EMACTIVE;
+ em->protocol = UDP;
em->bskt = idx;
+ em->delay = 10;
@-
All tables are prepended with a default tick bat.
It becomes the synchronization handle.
@@ -255,11 +255,6 @@
#ifdef _DEBUG_EMITTER_
mnstr_printf(EMout, "Instantiate a new emitter %d fields\n", i);
#endif
- em->status = EMLISTEN;
- (void)ret;
- if ( MT_create_thread(&em->pid, (void (*)(void *))EMstartThread, em,
MT_THR_DETACHED) != 0) {
- throw(MAL, "emitter.start", "Emitter initiation failed");
- }
return MAL_SUCCEED;
}
@@ -291,14 +286,18 @@
if (em == NULL)
throw(MAL, "emitter.resume", "Emitter not defined");
if (em->status == EMLISTEN)
- throw(MAL, "emitter.resume", "Emitter already started");
+ throw(MAL, "emitter.resume", "Emitter running, stop it first");
#ifdef _DEBUG_EMITTER_
mnstr_printf(EMout, "resume an emitter\n");
#endif
+ em->status = EMLISTEN;
+ (void)ret;
+ if ( MT_create_thread(&em->pid, (void (*)(void *))EMstartThread, em,
MT_THR_DETACHED) != 0)
+ throw(MAL, "emitter.start", "Emitter initiation failed");
return MAL_SUCCEED;
}
-str EMdrop(int *ret, str *nme)
+str EMstop(int *ret, str *nme)
{
Emitter em, rb;
@@ -328,7 +327,7 @@
Emitter r,o;
for( r= emAnchor; r; r = o){
o= r->nxt;
- EMdrop(ret, &r->name);
+ EMstop(ret, &r->name);
}
return MAL_SUCCEED;
}
@@ -409,7 +408,10 @@
bodyRestart:
/* create the actual channel */
- em->emitter = socket_wastream(em->newsockfd, em->name);
+ if ( em->protocol == UDP)
+ em->emitter = udp_wastream(em->host, em->port,
em->name);
+ else
+ em->emitter = socket_wastream(em->newsockfd, em->name);
if (em->emitter == NULL) {
perror("Emitter: Could not open stream");
mnstr_printf(EMout, "stream %s.%d.%s\n", em->host, em->port,
em->name);
@@ -483,10 +485,12 @@
mnstr_printf(GDKerr, em->table.error);
em->table.error = 0;
}
- }
+ } else
+ if ( em->delay)
+ MT_sleep_ms(em->delay);
}
/* writing failed, lets restart */
- if (em->protocol == EMPASSIVE) {
+ if (em->mode == EMPASSIVE) {
#ifdef _DEBUG_EMITTER_
mnstr_printf(EMout, "Restart the connection\n");
#endif
@@ -506,15 +510,9 @@
#ifdef _DEBUG_EMITTER_
mnstr_printf(EMout, "Emitter body %s started at %s:%d, servermode=%d\n",
- em->name, em->host, em->port, em->protocol);
+ em->name, em->host, em->port, em->mode);
#endif
- if (protocol == UDP) {
- em->emitter = udp_wastream(em->host, em->port, em->name);
- if (em->emitter == NULL) {
- perror("Emitter: Could not open stream");
- mnstr_printf(EMout, "stream %s.%d.%s\n", em->host,
em->port, em->name);
- return MAL_SUCCEED;
- }
+ if (em->mode == EMACTIVE) {
EMbody(em);
#ifdef _DEBUG_EMITTER_
mnstr_printf(EMout, "End of emitter thread\n");
@@ -523,7 +521,7 @@
}
/* Handling the TCP connection */
- if (em->protocol == EMPASSIVE &&
+ if (em->mode == EMPASSIVE &&
(em->error = socket_server_connect(&em->sockfd, em->port))) {
em->status = EMERROR;
mnstr_printf(EMout, "EMSTART THREAD: failed to start
server:%s\n", em->error);
@@ -531,7 +529,7 @@
}
do {
if (em->status == EMSTOP) return MAL_SUCCEED;
- if (em->protocol == EMPASSIVE) {
+ if (em->mode == EMPASSIVE) {
#ifdef _DEBUG_EMITTER_
mnstr_printf(EMout, "Emitter listens\n");
#endif
@@ -543,7 +541,7 @@
}
/* in server mode you should expect new connections */
- if (em->protocol == EMACTIVE) {
+ if (em->mode == EMACTIVE) {
if (MT_create_thread(&em->pid, (void (*)(void
*))EMbody, em, MT_THR_DETACHED) != 0) {
close_stream(em->emitter);
throw(MAL, "emitter.start", "Process creation
failed");
@@ -552,7 +550,7 @@
EMreconnect(em);
EMbody(em);
}
- } while (em->protocol == EMPASSIVE);
+ } while (em->mode == EMPASSIVE);
shutdown(em->newsockfd, SHUT_RDWR);
return MAL_SUCCEED;
}
_______________________________________________
Checkin-list mailing list
[email protected]
http://mail.monetdb.org/mailman/listinfo/checkin-list
