diff --git a/examples/twoway/client.c b/examples/twoway/client.c
index 3714abfeb1ff5fa2ff3ab830d3cc38f21f70ce8f..484b20edcabd2f7a2daef7b786d26846de61e9e9 100644
--- a/examples/twoway/client.c
+++ b/examples/twoway/client.c
@@ -85,6 +85,7 @@ int main(int argc, char *argv[])
struct labcomm_scheduler *scheduler;
struct labcomm_decoder *decoder;
struct labcomm_encoder *encoder;
+ struct labcomm_time *next;
int32_t i, j;
hostname = argv[1];
@@ -157,12 +158,10 @@ int main(int argc, char *argv[])
labcomm_encoder_register_types_B(encoder);
labcomm_encoder_register_types_Terminate(encoder);
- /* Sleep to make all remote types be known to introspecting and decimating
- wrappers, this is a HACK! */
- sleep(1);
err = labcomm_decoder_ioctl_types_Sum(decoder, SET_DECIMATION, 2);
err = labcomm_decoder_ioctl_types_Diff(decoder, SET_DECIMATION, 4);
+ next = labcomm_scheduler_now(scheduler);
for (i = 0 ; i < 4 ; i++) {
if (i == 2) {
labcomm_decoder_register_types_Product(decoder, handle_Product, NULL);
@@ -171,7 +170,8 @@ int main(int argc, char *argv[])
printf("\nA=%d B=%d: ", i, j);
labcomm_encode_types_A(encoder, &i);
labcomm_encode_types_B(encoder, &j);
- sleep(1);
+ labcomm_time_add_usec(next, 100000);
+ labcomm_scheduler_sleep(scheduler, next);
}
}
printf("\n");
diff --git a/examples/twoway/decimating.c b/examples/twoway/decimating.c
index 3b3cce3550f74838abf1dfbcff1e6f12a7426f4a..dbcc80cc2ed52f4c4019cbcec4c0d2c9dfe9aa00 100644
--- a/examples/twoway/decimating.c
+++ b/examples/twoway/decimating.c
@@ -35,11 +35,12 @@ struct decimating_private {
int encoder_initialized;
struct labcomm_reader_action_context reader_action_context;
struct labcomm_writer_action_context writer_action_context;
- LABCOMM_SIGNATURE_ARRAY_DEF(decimation,
+ LABCOMM_SIGNATURE_ARRAY_DEF(writer_decimation,
struct decimation {
int n;
int current;
});
+ LABCOMM_SIGNATURE_ARRAY_DEF(reader_decimation, int);
};
static void set_decimation(
@@ -51,7 +52,7 @@ static void set_decimation(
labcomm_scheduler_data_lock(decimating->scheduler);
decimation = LABCOMM_SIGNATURE_ARRAY_REF(decimating->memory,
- decimating->decimation,
+ decimating->writer_decimation,
struct decimation,
value->signature_index);
decimation->n = value->decimation;
@@ -87,34 +88,81 @@ static void send_set_decimation(void *arg)
labcomm_memory_free(memory, 1, msg);
}
+static void enqueue_decimation(struct decimating_private *decimating,
+ int remote_index,
+ int amount)
+{
+ struct send_set_decimation *msg;
+ msg = labcomm_memory_alloc(decimating->memory, 1, sizeof(*msg));
+ if (msg) {
+ msg->decimating = decimating;
+ msg->set_decimation.decimation = amount;
+ msg->set_decimation.signature_index = remote_index;
+
+ labcomm_scheduler_enqueue(decimating->scheduler, 0,
+ send_set_decimation, msg);
+ }
+}
+
+static int wrap_reader_start(
+ struct labcomm_reader *r,
+ struct labcomm_reader_action_context *action_context,
+ int local_index, int remote_index, struct labcomm_signature *signature,
+ void *value)
+{
+ struct decimating_private *decimating = action_context->context;
+
+ if (value == NULL) {
+ int *decimation, amount;
+
+ labcomm_scheduler_data_lock(decimating->scheduler);
+ decimation = LABCOMM_SIGNATURE_ARRAY_REF(decimating->memory,
+ decimating->reader_decimation,
+ int,
+ local_index);
+ amount = *decimation;
+ labcomm_scheduler_data_unlock(decimating->scheduler);
+ if (remote_index != 0 && amount != 0) {
+ enqueue_decimation(decimating, remote_index, amount);
+ }
+ }
+ return labcomm_reader_start(r, action_context->next,
+ local_index, remote_index, signature, value);
+}
+
static int wrap_reader_ioctl(
struct labcomm_reader *r,
struct labcomm_reader_action_context *action_context,
- int signature_index,
+ int local_index, int remote_index,
struct labcomm_signature *signature,
uint32_t action, va_list args)
{
struct decimating_private *decimating = action_context->context;
if (action == SET_DECIMATION) {
- struct send_set_decimation *msg;
-
- msg = labcomm_memory_alloc(decimating->memory, 1, sizeof(*msg));
- if (msg) {
- va_list va;
-
- va_copy(va, args);
- msg->decimating = decimating;
- msg->set_decimation.decimation = va_arg(va, int);
- msg->set_decimation.signature_index = signature_index;
- va_end(va);
-
- labcomm_scheduler_enqueue(decimating->scheduler, 0,
- send_set_decimation, msg);
+ va_list va;
+ int amount;
+ int *decimation;
+
+ va_copy(va, args);
+ amount = va_arg(va, int);
+ va_end(va);
+
+ labcomm_scheduler_data_lock(decimating->scheduler);
+ decimation = LABCOMM_SIGNATURE_ARRAY_REF(decimating->memory,
+ decimating->reader_decimation,
+ int,
+ local_index);
+ *decimation = amount;
+ labcomm_scheduler_data_unlock(decimating->scheduler);
+ if (remote_index) {
+ enqueue_decimation(decimating, remote_index, amount);
}
+
} else {
return labcomm_reader_ioctl(r, action_context->next,
- signature_index, signature, action, args);
+ local_index, remote_index, signature,
+ action, args);
}
return 0;
}
@@ -122,7 +170,7 @@ static int wrap_reader_ioctl(
struct labcomm_reader_action decimating_reader_action = {
.alloc = wrap_reader_alloc,
.free = NULL,
- .start = NULL,
+ .start = wrap_reader_start,
.end = NULL,
.fill = NULL,
.ioctl = wrap_reader_ioctl
@@ -161,7 +209,7 @@ static int wrap_writer_start(
labcomm_scheduler_data_lock(decimating->scheduler);
decimation = LABCOMM_SIGNATURE_ARRAY_REF(decimating->memory,
- decimating->decimation,
+ decimating->writer_decimation,
struct decimation, index);
decimation->current++;
if (decimation->current < decimation->n) {
@@ -221,7 +269,8 @@ struct decimating *decimating_new(
result->error = error;
result->memory = memory;
result->scheduler = scheduler;
- LABCOMM_SIGNATURE_ARRAY_INIT(result->decimation, struct decimation);
+ LABCOMM_SIGNATURE_ARRAY_INIT(result->writer_decimation, struct decimation);
+ LABCOMM_SIGNATURE_ARRAY_INIT(result->reader_decimation, int);
goto out_ok;
diff --git a/examples/twoway/server.c b/examples/twoway/server.c
index a8f38061ae41560e043ec7f9215deecf101bd163..3a34efa464ab06c19b312a0434d70ea9f10c413a 100644
--- a/examples/twoway/server.c
+++ b/examples/twoway/server.c
@@ -20,6 +20,7 @@
*/
#include <arpa/inet.h>
+#include <linux/tcp.h>
#include <errno.h>
#include <pthread.h>
#include <stdlib.h>
@@ -61,10 +62,6 @@ static void handle_B(int32_t *value, void *context)
int status;
client->B = *value;
- client->Sum = client->A + client->B;
- client->Diff = client->A - client->B;
- labcomm_encode_types_Sum(client->encoder, &client->Sum);
- labcomm_encode_types_Diff(client->encoder, &client->Diff);
status = labcomm_encoder_ioctl_types_Product(client->encoder, HAS_SIGNATURE);
switch (status) {
case introspecting_unregistered:
@@ -77,6 +74,10 @@ static void handle_B(int32_t *value, void *context)
default:
break;
}
+ client->Sum = client->A + client->B;
+ client->Diff = client->A - client->B;
+ labcomm_encode_types_Sum(client->encoder, &client->Sum);
+ labcomm_encode_types_Diff(client->encoder, &client->Diff);
}
static void handle_Terminate(types_Terminate *value, void *context)
@@ -183,6 +184,7 @@ int main(int argc, char *argv[])
}
while (1) {
struct client *client;
+ int nodelay;
client = malloc(sizeof(*client));
if (client == NULL) {
@@ -198,6 +200,8 @@ int main(int argc, char *argv[])
result = errno;
goto failed_to_accept;
}
+ nodelay = 1;
+ setsockopt(client->fd, IPPROTO_TCP, TCP_NODELAY, &nodelay, sizeof(nodelay));
pthread_create(&client->main_thread, NULL, run_client, client);
}
diff --git a/lib/c/labcomm.c b/lib/c/labcomm.c
index 1721289e4b7b7451f57e0d8dda8c50ffdb7e1d38..0e983b10957000f0833c77bc80faf54b1aa53dbd 100644
--- a/lib/c/labcomm.c
+++ b/lib/c/labcomm.c
@@ -85,11 +85,12 @@ int labcomm_reader_fill(struct labcomm_reader *r,
int labcomm_reader_ioctl(struct labcomm_reader *r,
struct labcomm_reader_action_context *action_context,
- int index,
+ int local_index, int remote_index,
struct labcomm_signature *signature,
uint32_t ioctl_action, va_list args)
{
- UNWRAP(ioctl, r, action_context, index, signature, ioctl_action, args);
+ UNWRAP(ioctl, r, action_context,
+ local_index, remote_index, signature, ioctl_action, args);
}
int labcomm_writer_alloc(struct labcomm_writer *w,
diff --git a/lib/c/labcomm_decoder.c b/lib/c/labcomm_decoder.c
index 7d657d7f0a103dd0f47ac9b870c459dbea390054..9a82859c2ba5046338f153118778eca52be2a30d 100644
--- a/lib/c/labcomm_decoder.c
+++ b/lib/c/labcomm_decoder.c
@@ -175,9 +175,11 @@ static int decode_typedef_or_sample(struct labcomm_decoder *d, int kind)
.pos = 0,
.error = 0,
};
- struct labcomm_signature signature;
- int remote_index, err;
-
+ struct labcomm_signature signature, *local_signature;
+ int remote_index, local_index, err;
+
+ local_signature = NULL;
+ local_index = 0;
labcomm_writer_alloc(&writer, writer.action_context, "");
labcomm_writer_start(&writer, writer.action_context, 0, NULL, NULL);
remote_index = labcomm_read_packed32(d->reader); //int
@@ -219,6 +221,8 @@ static int decode_typedef_or_sample(struct labcomm_decoder *d, int kind)
bcmp((void*)s->signature->signature, (void*)signature.signature,
signature.size) == 0) {
s->remote_index = remote_index;
+ local_signature = s->signature;
+ local_index = i;
remote_to_local = LABCOMM_SIGNATURE_ARRAY_REF(d->memory,
d->remote_to_local, int,
remote_index);
@@ -228,6 +232,12 @@ static int decode_typedef_or_sample(struct labcomm_decoder *d, int kind)
}
}
labcomm_scheduler_data_unlock(d->scheduler);
+ if (local_signature) {
+ labcomm_reader_start(d->reader, d->reader->action_context,
+ local_index, remote_index, local_signature,
+ NULL);
+ labcomm_reader_end(d->reader, d->reader->action_context);
+ }
}
#if 0
if (! entry) {
@@ -344,7 +354,7 @@ int labcomm_decoder_ioctl(struct labcomm_decoder *d,
va_start(va, action);
result = labcomm_reader_ioctl(d->reader,
d->reader->action_context,
- 0, NULL, action, va);
+ 0, 0, NULL, action, va);
va_end(va);
return result;
}
@@ -363,12 +373,9 @@ int labcomm_internal_decoder_ioctl(struct labcomm_decoder *d,
struct sample_entry,
local_index)->remote_index;
labcomm_scheduler_data_unlock(d->scheduler);
- if (remote_index == 0) {
- result = -EAGAIN;
- } else {
- result = labcomm_reader_ioctl(d->reader, d->reader->action_context,
- remote_index, signature, action, va);
- }
+ result = labcomm_reader_ioctl(d->reader, d->reader->action_context,
+ local_index, remote_index,
+ signature, action, va);
return result;
}
diff --git a/lib/c/labcomm_private.h b/lib/c/labcomm_private.h
index e953f02a03e117645fbce7642042e079f58ee88b..6ea5b4379eab3697708c820531b8b39af3c13265 100644
--- a/lib/c/labcomm_private.h
+++ b/lib/c/labcomm_private.h
@@ -113,8 +113,13 @@ struct labcomm_reader_action {
*/
int (*free)(struct labcomm_reader *r,
struct labcomm_reader_action_context *action_context);
- /* 'start' is called right after a sample has arrived. In the case of
- a sample or typedef, 'value' == NULL.
+ /* 'start' is called at the following instances:
+ 1. When a sample is registered
+ (local_index != 0, remote_index == 0, value == NULL)
+ 2. When a sample definition is received
+ (local_index != 0, remote_index != 0, value == NULL)
+ 3. When a sample is received
+ (local_index != 0, remote_index != 0, value != NULL)
*/
int (*start)(struct labcomm_reader *r,
struct labcomm_reader_action_context *action_context,
@@ -127,7 +132,8 @@ struct labcomm_reader_action {
struct labcomm_reader_action_context *action_context);
int (*ioctl)(struct labcomm_reader *r,
struct labcomm_reader_action_context *action_context,
- int index, struct labcomm_signature *signature,
+ int local_index, int remote_index,
+ struct labcomm_signature *signature,
uint32_t ioctl_action, va_list args);
};
@@ -165,7 +171,8 @@ int labcomm_reader_fill(struct labcomm_reader *r,
struct labcomm_reader_action_context *action_context);
int labcomm_reader_ioctl(struct labcomm_reader *r,
struct labcomm_reader_action_context *action_context,
- int index, struct labcomm_signature *signature,
+ int local_index, int remote_index,
+ struct labcomm_signature *signature,
uint32_t ioctl_action, va_list args);
/*