Mirrors/VILLASnode
1
0
Fork 0
mirror of https://git.rwth-aachen.de/acs/public/villas/node/ synced 2025-03-16 00:00:02 +01:00
Code Issues Releases Wiki Activity

Merge branch 'fix-shmem' into fix-queue-signalled

Browse source
This commit is contained in:
Georg Reinke 2017-05-12 13:12:30 +02:00
commit 434504b788
3 changed files with 12 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
include/villas/sample.h
View file

@ -78,7 +78,7 @@ struct sample {
};
/** Request \p cnt samples from memory pool \p p and initialize them.
* This will leave the reference count of the sample to zero.
* The reference count will already be set to 1.
* Use the sample_get() function to increase it. */
int sample_alloc(struct pool *p, struct sample *smps[], int cnt);

15
lib/nodes/shmem.c
View file

@ -33,6 +33,7 @@
#include "nodes/shmem.h"
#include "plugin.h"
#include "shmem.h"
#include "timing.h"
#include "utils.h"
int shmem_parse(struct node *n, config_setting_t *cfg)
@ -159,8 +160,9 @@ int shmem_read(struct node *n, struct sample *smps[], unsigned cnt)
int ret, recv;
recv = shm->polling ? queue_pull_many(&shm->shared->in.q, (void**) smps, cnt)
: queue_signalled_pull_many(&shm->shared->in.qs, (void**) smps, cnt);
struct sample *shared_smps[cnt];
recv = shm->polling ? queue_pull_many(&shm->shared->in.q, (void**) shared_smps, cnt)
: queue_signalled_pull_many(&shm->shared->in.qs, (void**) shared_smps, cnt);
if (recv <= 0)
return recv;
@ -168,8 +170,13 @@ int shmem_read(struct node *n, struct sample *smps[], unsigned cnt)
/* Check if remote process is still running */
ret = atomic_load_explicit(&shm->shared->ext_stopped, memory_order_relaxed);
if (ret)
return ret;
return -1;
sample_copy_many(smps, shared_smps, recv);
sample_put_many(shared_smps, recv);
struct timespec ts_recv = time_now();
for (int i = 0; i < recv; i++)
smps[i]->ts.received = ts_recv;
return recv;
}
@ -196,8 +203,6 @@ int shmem_write(struct node *n, struct sample *smps[], unsigned cnt)
memcpy(shared_smps[i]->data, smps[i]->data, SAMPLE_DATA_LEN(len));
shared_smps[i]->length = len;
sample_get(shared_smps[i]);
}
if (atomic_load_explicit(&shm->shared->ext_stopped, memory_order_relaxed)) {

2
lib/web.c
View file

@ -223,11 +223,11 @@ int web_start(struct web *w)
error("WebSocket: failed to initialize server");
}
w->state = STATE_STARTED;
ret = pthread_create(&w->thread, NULL, worker, w);
if (ret)
error("Failed to start Web worker");
w->state = STATE_STARTED;
return ret;
}