aboutsummaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
authorMattias Andrée <maandree@operamail.com>2015-05-17 14:10:11 +0200
committerMattias Andrée <maandree@operamail.com>2015-05-17 14:10:11 +0200
commit92292702763f53a099a3dddf6212ad0bc91c5388 (patch)
tree864630cca661d8f4d369782b6e442998193401ff /src
parentbump version (diff)
downloadbus-92292702763f53a099a3dddf6212ad0bc91c5388.tar.gz
bus-92292702763f53a099a3dddf6212ad0bc91c5388.tar.bz2
bus-92292702763f53a099a3dddf6212ad0bc91c5388.tar.xz
implemented timed write, read and poll
Signed-off-by: Mattias Andrée <maandree@operamail.com>
Diffstat (limited to 'src')
-rw-r--r--src/bus.c185
1 files changed, 178 insertions, 7 deletions
diff --git a/src/bus.c b/src/bus.c
index 06c708a..516def1 100644
--- a/src/bus.c
+++ b/src/bus.c
@@ -22,6 +22,7 @@
* DEALINGS IN THE SOFTWARE.
*/
#define _XOPEN_SOURCE 700
+#define _GNU_SOURCE
#include "bus.h"
#include <stdlib.h>
@@ -119,7 +120,7 @@
semaphore_op(bus, semaphore, +1, flags)
/**
- * Wait for the value of a semphore to become 0
+ * Wait for the value of a semaphore to become 0
*
* @param bus:const bus_t * The bus
* @param semaphore:int The index of the semaphore, `S`, `W`, `X` or `Q`
@@ -130,6 +131,43 @@
semaphore_op(bus, semaphore, 0, flags)
/**
+ * Decrease the value of a semaphore by 1
+ *
+ * @param bus:const bus_t * The bus
+ * @param semaphore:int The index of the semaphore, `S`, `W`, `X` or `Q`
+ * @param flags:int `SEM_UNDO` if the action should be undone when the program exits,
+ * `IPC_NOWAIT` if the action should fail if it would block
+ * @param timeout:const struct timespec * The amount of time to wait before failing
+ * @return :int 0 on success, -1 on error
+ */
+#define acquire_semaphore_timed(bus, semaphore, flags, timeout) \
+ semaphore_op_timed(bus, semaphore, -1, flags, timeout)
+
+/**
+ * Increase the value of a semaphore by 1
+ *
+ * @param bus:const bus_t * The bus
+ * @param semaphore:int The index of the semaphore, `S`, `W`, `X` or `Q`
+ * @param flags:int `SEM_UNDO` if the action should be undone when the program exits
+ * @param timeout:const struct timespec * The amount of time to wait before failing
+ * @return :int 0 on success, -1 on error
+ */
+#define release_semaphore_timed(bus, semaphore, flags, timeout) \
+ semaphore_op_timed(bus, semaphore, +1, flags, timeout)
+
+/**
+ * Wait for the value of a semaphore to become 0
+ *
+ * @param bus:const bus_t * The bus
+ * @param semaphore:int The index of the semaphore, `S`, `W`, `X` or `Q`
+ * @param flags:int `IPC_NOWAIT` if the action should fail if it would block
+ * @param timeout:const struct timespec * The amount of time to wait before failing
+ * @return :int 0 on success, -1 on error
+ */
+#define zero_semaphore_timed(bus, semaphore, flags, timeout) \
+ semaphore_op_timed(bus, semaphore, 0, flags, timeout)
+
+/**
* Open the semaphore array
*
* @param bus:const bus_t * The bus
@@ -150,6 +188,21 @@
/**
+ * Set `delta` to the convertion of `timeout` from absolute to relative time,
+ * measured in the clock whose ID is specified by `clockid`
+ *
+ * @scope timeout:struct timespec Output variable for relative time
+ * @scope timeout:const struct timespec * The absolute time
+ * @scope clockid:clockid_t The clock time is measured
+ */
+#define DELTA \
+ do { \
+ if (absolute_time_to_delta_time(&delta, timeout, clockid) < 0) goto fail; \
+ else if ((delta.tv_sec < 0) || (delta.tv_nsec < 0)) { errno = EAGAIN; goto fail; } \
+ } while (0)
+
+
+/**
* If `flags & (bus_flag)`, this macro evalutes to `sys_flag`,
* otherwise this macro evalutes to 0.
*/
@@ -319,6 +372,27 @@ semaphore_op(const bus_t *bus, int semaphore, int delta, int flags)
/**
+ * Increase or decrease the value of a semaphore, or wait the it to become 0
+ *
+ * @param bus Bus information
+ * @param semaphore The index of the semaphore, `S`, `W`, `X` or `Q`
+ * @param delta The adjustment to make to the semaphore's value, 0 to wait for it to become 0
+ * @param flags `SEM_UNDO` if the action should be undone when the program exits
+ * @param timeout The amount of time to wait before failing
+ * @return 0 on success, -1 on error
+ */
+static int
+semaphore_op_timed(const bus_t *bus, int semaphore, int delta, int flags, const struct timespec *timeout)
+{
+ struct sembuf op;
+ op.sem_num = (unsigned short)semaphore;
+ op.sem_op = (short)delta;
+ op.sem_flg = (short)flags;
+ return semtimedop(bus->sem_id, &op, (size_t)1, timeout);
+}
+
+
+/**
* Set the value of a semaphore
*
* @param bus Bus information
@@ -722,10 +796,41 @@ fail:
int bus_write_timed(const bus_t *bus, const char *message,
const struct timespec *timeout, clockid_t clockid)
{
- /* TODO bus_write_timed */
+ int saved_errno;
+#ifndef BUS_SEMAPHORES_ARE_SYNCHRONOUS
+ int state = 0;
+#endif
+ struct timespec delta;
if (!timeout)
return bus_write(bus, message, 0);
- (void) bus, (void) message, (void) timeout, (void) clockid;
+
+ DELTA;
+ if (acquire_semaphore_timed(bus, X, SEM_UNDO, &delta) == -1)
+ return -1;
+ DELTA;
+ t(zero_semaphore_timed(bus, W, 0, &delta));
+ write_shared_memory(bus, message);
+#ifndef BUS_SEMAPHORES_ARE_SYNCHRONOUS
+ t(release_semaphore(bus, N, SEM_UNDO)); state++;
+#endif
+ t(write_semaphore(bus, Q, 0));
+ t(zero_semaphore(bus, S, 0));
+#ifndef BUS_SEMAPHORES_ARE_SYNCHRONOUS
+ t(acquire_semaphore(bus, N, SEM_UNDO)); state--;
+#endif
+ t(release_semaphore(bus, X, SEM_UNDO));
+ return 0;
+
+fail:
+ saved_errno = errno;
+#ifndef BUS_SEMAPHORES_ARE_SYNCHRONOUS
+ if (state > 0)
+ acquire_semaphore(bus, N, SEM_UNDO);
+#endif
+ release_semaphore(bus, X, SEM_UNDO);
+ errno = saved_errno;
+ return -1;
+
}
@@ -822,10 +927,46 @@ done:
int bus_read_timed(const bus_t *bus, int (*callback)(const char *message, void *user_data),
void *user_data, const struct timespec *timeout, clockid_t clockid)
{
- /* TODO bus_read_timed */
+ int r, state = 0, saved_errno;
+ struct timespec delta;
if (!timeout)
return bus_read(bus, callback, user_data);
- (void) bus, (void) callback, (void) user_data, (void) timeout, (void) clockid;
+
+ DELTA;
+ if (release_semaphore_timed(bus, S, SEM_UNDO, &delta) == -1)
+ return -1;
+ t(r = callback(NULL, user_data));
+ if (!r) goto done;
+ for (;;) {
+ DELTA;
+ t(release_semaphore_timed(bus, Q, 0, &delta));
+ DELTA;
+ t(zero_semaphore_timed(bus, Q, 0, &delta));
+ t(r = callback(bus->message, user_data));
+ if (!r) goto done;
+ t(release_semaphore(bus, W, SEM_UNDO)); state++;
+ t(acquire_semaphore(bus, S, SEM_UNDO)); state++;
+ t(zero_semaphore(bus, S, 0));
+#ifndef BUS_SEMAPHORES_ARE_SYNCHRONOUS_ME_HARDER
+ t(zero_semaphore(bus, N, 0));
+#endif
+ t(release_semaphore(bus, S, SEM_UNDO)); state--;
+ t(acquire_semaphore(bus, W, SEM_UNDO)); state--;
+ }
+
+fail:
+ saved_errno = errno;
+ if (state > 1)
+ release_semaphore(bus, S, SEM_UNDO);
+ if (state > 0)
+ acquire_semaphore(bus, W, SEM_UNDO);
+ acquire_semaphore(bus, S, SEM_UNDO);
+ errno = saved_errno;
+ return -1;
+
+done:
+ t(acquire_semaphore(bus, S, SEM_UNDO));
+ return 0;
}
@@ -943,10 +1084,40 @@ fail:
*/
const char *bus_poll_timed(bus_t *bus, const struct timespec *timeout, clockid_t clockid)
{
- /* TODO bus_poll_timed */
+ int state = 0, saved_errno;
+ struct timespec delta;
if (!timeout)
return bus_poll(bus);
- (void) bus, (void) timeout, (void) clockid;
+
+ if (!bus->first_poll) {
+ t(release_semaphore(bus, W, SEM_UNDO)); state++;
+ t(acquire_semaphore(bus, S, SEM_UNDO)); state++;
+ t(zero_semaphore(bus, S, 0));
+#ifndef BUS_SEMAPHORES_ARE_SYNCHRONOUS_ME_HARDER
+ t(zero_semaphore(bus, N, 0));
+#endif
+ t(release_semaphore(bus, S, SEM_UNDO)); state--;
+ t(acquire_semaphore(bus, W, SEM_UNDO)); state--;
+ } else {
+ bus->first_poll = 0;
+ }
+ state--;
+ DELTA;
+ t(release_semaphore_timed(bus, Q, 0, &delta));
+ DELTA;
+ t(zero_semaphore_timed(bus, Q, 0, &delta));
+ return bus->message;
+
+fail:
+ saved_errno = errno;
+ if (state > 1)
+ release_semaphore(bus, S, SEM_UNDO);
+ if (state > 0)
+ acquire_semaphore(bus, W, SEM_UNDO);
+ if (state < 0)
+ bus->first_poll = 1;
+ errno = saved_errno;
+ return NULL;
}