Introduce pass_object_as_shm().
[paraslash.git] / afs.c
diff --git a/afs.c b/afs.c
index 20f2455e30a3d1fe940d84d9d5616fd97e5ee30d..d41a518d791f9908b0230beda99849f0a8f4c3e9 100644 (file)
--- a/afs.c
+++ b/afs.c
@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2007 Andre Noll <maan@systemlinux.org>
+ * Copyright (C) 2007-2008 Andre Noll <maan@systemlinux.org>
  *
  * Licensed under the GPL v2. For licencing details see COPYING.
  */
@@ -148,7 +148,7 @@ struct callback_result {
  * \param result Callback result will be stored here.
  *
  * This function creates a shared memory area, copies the buffer pointed to by
- * query to that area and notifies the afs process that \a f should be
+ * \a query to that area and notifies the afs process that \a f should be
  * called ASAP.
  *
  * \return Negative, on errors, the return value of the callback function
@@ -157,14 +157,13 @@ struct callback_result {
  * \sa send_option_arg_callback_request(), send_standard_callback_request().
  */
 int send_callback_request(callback_function *f, struct osl_object *query,
-               struct osl_object *result)
+               callback_result_handler *result_handler,
+               void *private_result_data)
 {
        struct callback_query *cq;
-       struct callback_result *cr;
-       int ret, fd = -1, query_shmid, result_shmid;
+       int num_results = 0, ret, fd = -1, query_shmid, result_shmid;
        void *query_shm, *result_shm;
        char buf[sizeof(afs_socket_cookie) + sizeof(int)];
-       struct sockaddr_un unix_addr;
        size_t query_shm_size = sizeof(*cq);
 
        if (query)
@@ -189,50 +188,52 @@ int send_callback_request(callback_function *f, struct osl_object *query,
        *(uint32_t *) buf = afs_socket_cookie;
        *(int *) (buf + sizeof(afs_socket_cookie)) = query_shmid;
 
-       ret = get_stream_socket(PF_UNIX);
+       ret = create_remote_socket(conf.afs_socket_arg);
        if (ret < 0)
                goto out;
        fd = ret;
-       ret = init_unix_addr(&unix_addr, conf.afs_socket_arg);
-       if (ret < 0)
-               goto out;
-       ret = PARA_CONNECT(fd, &unix_addr);
-       if (ret < 0)
-               goto out;
        ret = send_bin_buffer(fd, buf, sizeof(buf));
        if (ret < 0)
                goto out;
-       ret = recv_bin_buffer(fd, buf, sizeof(buf));
-       if (ret < 0)
-               goto out;
-       if (ret != sizeof(int)) {
-               ret = -E_RECV;
-               goto out;
-       }
-       ret = *(int *) buf;
-       if (ret <= 0)
-               goto out;
-       result_shmid = ret;
-       ret = shm_attach(result_shmid, ATTACH_RO, &result_shm);
-       if (ret >= 0) {
-               assert(result);
-               cr = result_shm;
-               result->size = cr->result_size;
-               result->data = para_malloc(result->size);
-               memcpy(result->data, result_shm + sizeof(*cr), result->size);
-               ret = shm_detach(result_shm);
+       for (;;) {
+               ret = recv_bin_buffer(fd, buf, sizeof(int));
+               if (ret <= 0)
+                       goto out;
+               if (ret != sizeof(int)) {
+                       ret = -E_AFS_SHORT_READ;
+                       goto out;
+               }
+               ret = *(int *) buf;
+               if (ret <= 0)
+                       goto out;
+               result_shmid = ret;
+               ret = shm_attach(result_shmid, ATTACH_RO, &result_shm);
+               if (ret >= 0) {
+                       struct callback_result *cr = result_shm;
+                       struct osl_object result;
+                       num_results++;
+                       result.size = cr->result_size;
+                       result.data = result_shm + sizeof(*cr);
+                       if (result.size) {
+                               assert(result_handler);
+                               ret = result_handler(&result, private_result_data);
+                               if (shm_detach(result_shm) < 0)
+                                       PARA_ERROR_LOG("can not detach result\n");
+                       }
+               } else
+                       PARA_ERROR_LOG("attach result failed: %d\n", ret);
+               if (shm_destroy(result_shmid) < 0)
+                       PARA_ERROR_LOG("destroy result failed\n");
                if (ret < 0)
-                       PARA_ERROR_LOG("can not detach result\n");
-       } else
-               PARA_ERROR_LOG("attach result failed: %d\n", ret);
-       if (shm_destroy(result_shmid) < 0)
-               PARA_ERROR_LOG("destroy result failed\n");
-       ret = 1;
+                       break;
+       }
 out:
        if (shm_destroy(query_shmid) < 0)
                PARA_ERROR_LOG("%s\n", "shm destroy error");
        if (fd >= 0)
                close(fd);
+       if (ret >= 0)
+               ret = num_results;
 //     PARA_DEBUG_LOG("callback_ret: %d\n", ret);
        return ret;
 }
@@ -255,7 +256,8 @@ out:
  */
 int send_option_arg_callback_request(struct osl_object *options,
                int argc,  char * const * const argv, callback_function *f,
-               struct osl_object *result)
+               callback_result_handler *result_handler,
+               void *private_result_data)
 {
        char *p;
        int i, ret;
@@ -273,7 +275,8 @@ int send_option_arg_callback_request(struct osl_object *options,
                strcpy(p, argv[i]); /* OK */
                p += strlen(argv[i]) + 1;
        }
-       ret = send_callback_request(f, &query, result);
+       ret = send_callback_request(f, &query, result_handler,
+               private_result_data);
        free(query.data);
        return ret;
 }
@@ -293,9 +296,11 @@ int send_option_arg_callback_request(struct osl_object *options,
  * send_option_arg_callback_request().
  */
 int send_standard_callback_request(int argc,  char * const * const argv,
-               callback_function *f, struct osl_object *result)
+               callback_function *f, callback_result_handler *result_handler,
+               void *private_result_data)
 {
-       return send_option_arg_callback_request(NULL, argc, argv, f, result);
+       return send_option_arg_callback_request(NULL, argc, argv, f, result_handler,
+               private_result_data);
 }
 
 static int action_if_pattern_matches(struct osl_row *row, void *data)
@@ -414,7 +419,8 @@ static int fd2buf(int fd, unsigned max_size, struct osl_object *obj)
  * send_callback_request() otherwise.
  */
 int stdin_command(int fd, struct osl_object *arg_obj, callback_function *f,
-               unsigned max_len, struct osl_object *result)
+               unsigned max_len, callback_result_handler *result_handler,
+               void *private_result_data)
 {
        struct osl_object query, stdin_obj;
        int ret;
@@ -430,7 +436,7 @@ int stdin_command(int fd, struct osl_object *arg_obj, callback_function *f,
        memcpy(query.data, arg_obj->data, arg_obj->size);
        memcpy((char *)query.data + arg_obj->size, stdin_obj.data, stdin_obj.size);
        free(stdin_obj.data);
-       ret = send_callback_request(f, &query, result);
+       ret = send_callback_request(f, &query, result_handler, private_result_data);
        free(query.data);
        return ret;
 }
@@ -486,17 +492,27 @@ int open_next_audio_file(void)
        struct audio_file_data afd;
        int ret, shmid;
        char buf[8];
-
+       long score;
+again:
        PARA_NOTICE_LOG("getting next audio file\n");
-       ret = score_get_best(&aft_row, &afd.score);
-       if (ret < 0)
-               return ret;
-       ret = open_and_update_audio_file(aft_row, &afd);
-       if (ret < 0)
-               return ret;
+       ret = score_get_best(&aft_row, &score);
+       if (ret < 0) {
+               PARA_ERROR_LOG("%s\n", para_strerror(-ret));
+               goto no_admissible_files;
+       }
+       ret = open_and_update_audio_file(aft_row, score, &afd);
+       if (ret < 0) {
+               PARA_ERROR_LOG("%s\n", para_strerror(-ret));
+               ret = score_delete(aft_row);
+               if (ret < 0) {
+                       PARA_ERROR_LOG("%s\n", para_strerror(-ret));
+                       goto no_admissible_files;
+               }
+               goto again;
+       }
        shmid = ret;
        if (!write_ok(server_socket)) {
-               PARA_EMERG_LOG("afs_socket not writable\n");
+               ret = -E_AFS_SOCKET;
                goto destroy;
        }
        *(uint32_t *)buf = NEXT_AUDIO_FILE;
@@ -505,10 +521,13 @@ int open_next_audio_file(void)
        close(afd.fd);
        if (ret >= 0)
                return ret;
-       PARA_ERROR_LOG("%s\n", PARA_STRERROR(-ret));
 destroy:
        shm_destroy(shmid);
        return ret;
+no_admissible_files:
+       *(uint32_t *)buf = NO_ADMISSIBLE_FILES;
+       *(uint32_t *)(buf + 4) = (uint32_t)0;
+       return send_bin_buffer(server_socket, buf, 8);
 }
 
 /* Never fails if arg == NULL */
@@ -521,14 +540,14 @@ static int activate_mood_or_playlist(char *arg, int *num_admissible)
                ret = change_current_mood(NULL); /* always successful */
                mode = PLAY_MODE_MOOD;
        } else {
-               if (!strncmp(arg, "p:", 2)) {
+               if (!strncmp(arg, "p/", 2)) {
                        ret = playlist_open(arg + 2);
                        mode = PLAY_MODE_PLAYLIST;
-               } else if (!strncmp(arg, "m:", 2)) {
+               } else if (!strncmp(arg, "m/", 2)) {
                        ret = change_current_mood(arg + 2);
                        mode = PLAY_MODE_MOOD;
                } else
-                       ret = -E_AFS_SYNTAX;
+                       return -E_AFS_SYNTAX;
                if (ret < 0)
                        return ret;
        }
@@ -570,7 +589,7 @@ static int com_select_callback(const struct osl_object *query,
                playlist_close();
        ret = activate_mood_or_playlist(arg, &num_admissible);
        if (ret < 0) {
-               para_printf(&pb, "%s\n", PARA_STRERROR(-ret));
+               para_printf(&pb, "%s\n", para_strerror(-ret));
                para_printf(&pb, "switching back to %s\n", current_mop?
                        current_mop : "dummy");
                ret = activate_mood_or_playlist(current_mop, &num_admissible);
@@ -582,26 +601,28 @@ static int com_select_callback(const struct osl_object *query,
        para_printf(&pb, "activated %s (%d admissible files)\n", current_mop?
                current_mop : "dummy mood", num_admissible);
        result->data = pb.buf;
-       result->size = pb.size;
+       result->size = pb.offset;
        return 1;
 }
 
+int send_result(struct osl_object *result, void *private_result_data)
+{
+       int fd = *(int *)private_result_data;
+       if (!result->size)
+               return 1;
+       return send_bin_buffer(fd, result->data, result->size);
+}
+
 int com_select(int fd, int argc, char * const * const argv)
 {
-       int ret;
-       struct osl_object query, result;
+       struct osl_object query;
 
        if (argc != 2)
                return -E_AFS_SYNTAX;
        query.data = argv[1];
        query.size = strlen(argv[1]) + 1;
-       ret = send_callback_request(com_select_callback, &query,
-               &result);
-       if (ret > 0 && result.data && result.size) {
-               ret = send_va_buffer(fd, "%s", (char *)result.data);
-               free(result.data);
-       }
-       return ret;
+       return send_callback_request(com_select_callback, &query,
+               &send_result, &fd);
 }
 
 static void init_admissible_files(char *arg)
@@ -620,7 +641,7 @@ static int setup_command_socket_or_die(void)
        ret = create_local_socket(socket_name, &unix_addr,
                S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP | S_IWOTH);
        if (ret < 0) {
-               PARA_EMERG_LOG("%s: %s\n", PARA_STRERROR(-ret), socket_name);
+               PARA_EMERG_LOG("%s: %s\n", para_strerror(-ret), socket_name);
                exit(EXIT_FAILURE);
        }
        socket_fd = ret;
@@ -628,12 +649,13 @@ static int setup_command_socket_or_die(void)
                PARA_EMERG_LOG("can not listen on socket\n");
                exit(EXIT_FAILURE);
        }
-       ret = mark_fd_nonblock(socket_fd);
+       ret = mark_fd_nonblocking(socket_fd);
        if (ret < 0) {
                close(socket_fd);
                return ret;
        }
-       PARA_INFO_LOG("listening on socket %s (fd %d)\n", socket_name, ret);
+       PARA_INFO_LOG("listening on socket %s (fd %d)\n", socket_name,
+               socket_fd);
        return socket_fd;
 }
 
@@ -685,7 +707,7 @@ static int open_afs_tables(void)
                if (ret >= 0)
                        continue;
                PARA_ERROR_LOG("%s init: %s\n", afs_tables[i].name,
-                       PARA_STRERROR(-ret));
+                       para_strerror(-ret));
                break;
        }
        if (ret >= 0)
@@ -731,7 +753,7 @@ static void signal_post_select(struct sched *s, struct task *t)
        }
        t->ret = -E_AFS_SIGNAL;
 err:
-       PARA_NOTICE_LOG("%s\n", PARA_STRERROR(-t->ret));
+       PARA_NOTICE_LOG("%s\n", para_strerror(-t->ret));
        unregister_tasks();
 }
 
@@ -776,18 +798,47 @@ static void command_pre_select(struct sched *s, struct task *t)
        t->ret = 1;
 }
 
+int pass_object_as_shm(int fd, struct osl_object *obj)
+{
+       int ret, shmid;
+       void *shm;
+       struct callback_result *cr;
+
+       if (!obj->data || !obj->size)
+               return 0;
+       ret = shm_new(obj->size + sizeof(struct callback_result));
+       if (ret < 0)
+               return ret;
+       shmid = ret;
+       ret = shm_attach(shmid, ATTACH_RW, &shm);
+       if (ret < 0)
+               goto err;
+       cr = shm;
+       cr->result_size = obj->size;
+       memcpy(shm + sizeof(*cr), obj->data, obj->size);
+       ret = shm_detach(shm);
+       if (ret < 0)
+               goto err;
+       ret = send_bin_buffer(fd, (char *)&shmid, sizeof(int));
+       if (ret >= 0)
+               return ret;
+err:
+       if (shm_destroy(shmid) < 0)
+               PARA_ERROR_LOG("destroy result failed\n");
+       return ret;
+}
+
 /*
  * On errors, negative value is written to fd.
  * On success: If query produced a result, the result_shmid is written to fd.
  * Otherwise, zero is written.
  */
-static int call_callback(int fd, int query_shmid)
+static void call_callback(int fd, int query_shmid)
 {
-       void *query_shm, *result_shm;
+       void *query_shm;
        struct callback_query *cq;
-       struct callback_result *cr;
        struct osl_object query, result = {.data = NULL};
-       int result_shmid = -1, ret, ret2;
+       int ret;
 
        ret = shm_attach(query_shmid, ATTACH_RW, &query_shm);
        if (ret < 0)
@@ -796,39 +847,13 @@ static int call_callback(int fd, int query_shmid)
        query.data = (char *)query_shm + sizeof(*cq);
        query.size = cq->query_size;
        ret = cq->handler(&query, &result);
-       ret2 = shm_detach(query_shm);
-       if (ret2 < 0 && ret >= 0)
-               ret = ret2;
-       if (ret < 0)
-               goto out;
-       ret = 0;
-       if (!result.data || !result.size)
-               goto out;
-       ret = shm_new(result.size + sizeof(struct callback_result));
        if (ret < 0)
                goto out;
-       result_shmid = ret;
-       ret = shm_attach(result_shmid, ATTACH_RW, &result_shm);
-       if (ret < 0)
-               goto out;
-       cr = result_shm;
-       cr->result_size = result.size;
-       memcpy(result_shm + sizeof(*cr), result.data, result.size);
-       ret = shm_detach(result_shm);
-       if (ret < 0)
-               goto out;
-       ret = result_shmid;
+       ret = pass_object_as_shm(fd, &result);
 out:
        free(result.data);
-       ret2 = send_bin_buffer(fd, (char *)&ret, sizeof(int));
-       if (ret < 0 || ret2 < 0) {
-               if (result_shmid >= 0)
-                       if (shm_destroy(result_shmid) < 0)
-                               PARA_ERROR_LOG("destroy result failed\n");
-               if (ret >= 0)
-                       ret = ret2;
-       }
-       return ret;
+       if (ret < 0)
+               PARA_ERROR_LOG("%s\n", para_strerror(-ret));
 }
 
 static void execute_server_command(void)
@@ -838,15 +863,17 @@ static void execute_server_command(void)
 
        if (ret <= 0) {
                if (ret < 0)
-                       PARA_ERROR_LOG("%s\n", PARA_STRERROR(-ret));
+                       PARA_ERROR_LOG("%s\n", para_strerror(-ret));
                return;
        }
        buf[ret] = '\0';
        PARA_DEBUG_LOG("received: %s\n", buf);
        if (!strcmp(buf, "new")) {
                ret = open_next_audio_file();
-               if (ret < 0)
-                       PARA_ERROR_LOG("%s\n", PARA_STRERROR(-ret));
+               if (ret < 0) {
+                       PARA_EMERG_LOG("%s\n", para_strerror(-ret));
+                       unregister_tasks();
+               }
                return;
        }
        PARA_ERROR_LOG("unknown command\n");
@@ -861,7 +888,7 @@ static void execute_afs_command(int fd, uint32_t expected_cookie)
        int ret = recv_bin_buffer(fd, buf, sizeof(buf));
 
        if (ret < 0) {
-               PARA_NOTICE_LOG("%s\n", PARA_STRERROR(-ret));
+               PARA_NOTICE_LOG("%s\n", para_strerror(-ret));
                return;
        }
        if (ret != sizeof(buf)) {
@@ -881,7 +908,6 @@ static void execute_afs_command(int fd, uint32_t expected_cookie)
                        query_shmid);
                return;
        }
-       /* Ignore return value: Errors might be OK here. */
        call_callback(fd, query_shmid);
 }
 
@@ -917,13 +943,13 @@ static void command_post_select(struct sched *s, struct task *t)
                goto out;
        t->ret = para_accept(ct->fd, &unix_addr, sizeof(unix_addr));
        if (t->ret < 0) {
-               PARA_NOTICE_LOG("%s\n", PARA_STRERROR(-t->ret));
+               PARA_NOTICE_LOG("%s\n", para_strerror(-t->ret));
                goto out;
        }
        fd = t->ret;
-       t->ret = mark_fd_nonblock(fd);
+       t->ret = mark_fd_nonblocking(fd);
        if (t->ret < 0) {
-               PARA_NOTICE_LOG("%s\n", PARA_STRERROR(-t->ret));
+               PARA_NOTICE_LOG("%s\n", para_strerror(-t->ret));
                close(fd);
                goto out;
        }
@@ -971,11 +997,11 @@ __noreturn void afs_init(uint32_t cookie, int socket_fd)
        ret = open_afs_tables();
 
        if (ret < 0) {
-               PARA_EMERG_LOG("%s\n", PARA_STRERROR(-ret));
+               PARA_EMERG_LOG("%s\n", para_strerror(-ret));
                exit(EXIT_FAILURE);
        }
        server_socket = socket_fd;
-       ret = mark_fd_nonblock(server_socket);
+       ret = mark_fd_nonblocking(server_socket);
        if (ret < 0)
                exit(EXIT_FAILURE);
        PARA_INFO_LOG("server_socket: %d, afs_socket_cookie: %u\n",
@@ -984,9 +1010,9 @@ __noreturn void afs_init(uint32_t cookie, int socket_fd)
        register_tasks(cookie);
        s.default_timeout.tv_sec = 0;
        s.default_timeout.tv_usec = 999 * 1000;
-       ret = sched(&s);
+       ret = schedule(&s);
        if (ret < 0)
-               PARA_EMERG_LOG("%s\n", PARA_STRERROR(-ret));
+               PARA_EMERG_LOG("%s\n", para_strerror(-ret));
        close_afs_tables();
        exit(EXIT_FAILURE);
 }
@@ -1038,7 +1064,7 @@ int com_init(int fd, int argc, char * const * const argv)
                                return -E_BAD_TABLE_NAME;
                }
        }
-       ret = send_callback_request(create_tables_callback, &query, NULL);
+       ret = send_callback_request(create_tables_callback, &query, NULL, NULL);
        if (ret < 0)
                return ret;
        return send_va_buffer(fd, "successfully created afs table(s)\n");
@@ -1062,7 +1088,6 @@ int com_check(int fd, int argc, char * const * const argv)
 {
        unsigned flags = 0;
        int i, ret;
-       struct osl_object result;
 
        for (i = 1; i < argc; i++) {
                const char *arg = argv[i];
@@ -1091,37 +1116,19 @@ int com_check(int fd, int argc, char * const * const argv)
        if (!flags)
                flags = ~0U;
        if (flags & CHECK_AFT) {
-               ret = send_callback_request(aft_check_callback, NULL, &result);
+               ret = send_callback_request(aft_check_callback, NULL, send_result, &fd);
                if (ret < 0)
                        return ret;
-               if (ret > 0) {
-                       ret = send_buffer(fd, (char *) result.data);
-                       free(result.data);
-                       if (ret < 0)
-                               return ret;
-               }
        }
        if (flags & CHECK_PLAYLISTS) {
-               ret = send_callback_request(playlist_check_callback, NULL, &result);
+               ret = send_callback_request(playlist_check_callback, NULL, send_result, &fd);
                if (ret < 0)
                        return ret;
-               if (ret > 0) {
-                       ret = send_buffer(fd, (char *) result.data);
-                       free(result.data);
-                       if (ret < 0)
-                               return ret;
-               }
        }
        if (flags & CHECK_MOODS) {
-               ret = send_callback_request(mood_check_callback, NULL, &result);
+               ret = send_callback_request(mood_check_callback, NULL, send_result, &fd);
                if (ret < 0)
                        return ret;
-               if (ret > 0) {
-                       ret = send_buffer(fd, (char *) result.data);
-                       free(result.data);
-                       if (ret < 0)
-                               return ret;
-               }
        }
        return 1;
 }
@@ -1137,7 +1144,7 @@ void afs_event(enum afs_events event, struct para_buffer *pb,
                        continue;
                ret = t->event_handler(event, pb, data);
                if (ret < 0)
-                       PARA_CRIT_LOG("%s\n", PARA_STRERROR(-ret));
+                       PARA_CRIT_LOG("%s\n", para_strerror(-ret));
        }
 }