niklas/cyclonedds
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

explicitly enable batching in pubsub

Browse source 
Automatic batching is not yet implemented but: it dramatically enhances performance in throughput tests, and enabling it in the config breaks manual publication of data in pubsub. For the time being, it makes more sense to enable manual batching programmatically and force the data out when needed.

Signed-off-by: Erik Boasson <eb@ilities.com>
This commit is contained in:
Erik Boasson 2018-08-01 09:05:00 +02:00
parent 7d22a343fc
commit 296ef4de7d
1 changed files with 9 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

9
src/tools/pubsub/pubsub.c
View file

@ -1222,6 +1222,8 @@ static void pub_do_auto(const struct writerspec *spec) {
} }
if (++bi == spec->burstsize) { if (++bi == spec->burstsize) {
while (((ntot / spec->burstsize) / ((t - tfirst0) / 1e9 + 5e-3)) > spec->writerate && !termflag) { while (((ntot / spec->burstsize) / ((t - tfirst0) / 1e9 + 5e-3)) > spec->writerate && !termflag) {
/* FIXME: only doing this manually because batching is not yet implemented properly */
dds_write_flush(spec->wr);
dds_sleepfor(DDS_MSECS(10)); dds_sleepfor(DDS_MSECS(10));
t = dds_time(); t = dds_time();
} }
@ -1292,11 +1294,17 @@ static char *pub_do_nonarb(const struct writerspec *spec, uint32_t *seq) {
if (!accept_error(command, result)) if (!accept_error(command, result))
exit(1); exit(1);
} }
/* FIXME: only doing this manually because batching is not yet implemented properly */
dds_write_flush(spec->wr);
if (spec->dupwr && (result = fn(spec->dupwr, &d, tstamp)) != DDS_RETCODE_OK) { if (spec->dupwr && (result = fn(spec->dupwr, &d, tstamp)) != DDS_RETCODE_OK) {
printf ("%s %d(dup): error %d (%s)\n", get_write_operstr(command), k, (int) result, dds_err_str(result)); printf ("%s %d(dup): error %d (%s)\n", get_write_operstr(command), k, (int) result, dds_err_str(result));
if (!accept_error(command, result)) if (!accept_error(command, result))
exit(1); exit(1);
} }
if (spec->dupwr) {
/* FIXME: only doing this manually because batching is not yet implemented properly */
dds_write_flush(spec->wr);
}
d.seq++; d.seq++;
break; break;
} }
@ -2492,6 +2500,7 @@ int MAIN(int argc, char *argv[]) {
common_init(argv[0]); common_init(argv[0]);
set_systemid_env(); set_systemid_env();
dds_write_set_batch(true); // FIXME: hack (the global batching flag is a hack anyway)
{ {
char **ps = (char **) dds_alloc(sizeof(char *) * (argc - os_get_optind())); char **ps = (char **) dds_alloc(sizeof(char *) * (argc - os_get_optind()));