From 9e34b1334b93d89a02bae43b319569be81a81618 Mon Sep 17 00:00:00 2001 From: Mike Brady Date: Wed, 27 May 2020 15:02:49 +0100 Subject: [PATCH] Stop using home-brew non-blocking write. Instead, having opened the pipe non-blocking, change to blocking. --- audio_pipe.c | 32 +++++----------- common.c | 104 ++++++++++++++++---------------------------------- common.h | 2 +- rtsp.c | 105 ++++++++++++++++++++++++--------------------------- 4 files changed, 93 insertions(+), 150 deletions(-) diff --git a/audio_pipe.c b/audio_pipe.c index 2abef94b..a2b33349 100644 --- a/audio_pipe.c +++ b/audio_pipe.c @@ -42,7 +42,6 @@ static int fd = -1; char *pipename = NULL; -int warned = 0; static void start(__attribute__((unused)) int sample_rate, __attribute__((unused)) int sample_format) { @@ -51,37 +50,26 @@ static void start(__attribute__((unused)) int sample_rate, // we check that it's not a "real" error though. From the "man 2 open" page: // "ENXIO O_NONBLOCK | O_WRONLY is set, the named file is a FIFO, and no process has the FIFO // open for reading." - fd = open(pipename, O_WRONLY | O_NONBLOCK); - if ((fd == -1) && (errno != ENXIO) && (warned == 0)) { - char errorstring[1024]; - strerror_r(errno, (char *)errorstring, sizeof(errorstring)); - debug(1, "pipe: start -- error %d (\"%s\") opening the pipe named \"%s\".", errno, - (char *)errorstring, pipename); - warn("Error %d opening the pipe named \"%s\".", errno, pipename); - warned = 1; - } -} + + fd = try_to_open_pipe_for_writing(pipename); + } static int play(void *buf, int samples) { // if the file is not open, try to open it. char errorstring[1024]; if (fd == -1) { - fd = open(pipename, O_WRONLY | O_NONBLOCK); + fd = try_to_open_pipe_for_writing(pipename); } // if it's got a reader, write to it. if (fd > 0) { - int rc = non_blocking_write(fd, buf, samples * 4); - if ((rc < 0) && (warned == 0)) { + //int rc = non_blocking_write(fd, buf, samples * 4); + int rc = write(fd, buf, samples * 4); + if ((rc < 0) && (errno != EPIPE)) { strerror_r(errno, (char *)errorstring, 1024); - warn("Error %d writing to the pipe named \"%s\": \"%s\".", errno, pipename, errorstring); - warned = 1; + debug(1, "audio_pip play: error %d writing to the pipe named \"%s\": \"%s\".", errno, pipename, errorstring); } - } else if ((fd == -1) && (errno != ENXIO) && (warned == 0)) { - strerror_r(errno, (char *)errorstring, 1024); - warn("Error %d opening the pipe named \"%s\": \"%s\".", errno, pipename, errorstring); - warned = 1; } - return warned; + return 0; } static void stop(void) { @@ -89,7 +77,7 @@ static void stop(void) { } static int init(int argc, char **argv) { - debug(1, "pipe init"); + // debug(1, "pipe init"); // const char *str; // int value; // double dvalue; diff --git a/common.c b/common.c index 43bb41af..74f89869 100644 --- a/common.c +++ b/common.c @@ -40,6 +40,7 @@ #include #include #include +#include #ifdef COMPILE_FOR_OSX #include @@ -1049,80 +1050,39 @@ uint64_t get_absolute_time_in_ns() { return time_now_ns; } -ssize_t non_blocking_write_with_timeout(int fd, const void *buf, size_t count, uint64_t timeout) { - // try to write to this non-blocking file descriptor - // wait for up to the timeout for it to be possible to write something. - // exit immediately if there is no connection at the other end - // timeout is in nanoseconds - void *ibuf = (void *)buf; - size_t bytes_remaining = count; - ssize_t rc = 0; - uint64_t end_time = get_absolute_time_in_ns() + timeout; - while ((bytes_remaining) && (rc == 0)) { - rc = write(fd, ibuf, bytes_remaining); - if (rc >= 0) { - bytes_remaining = bytes_remaining - rc; - ibuf += rc; - rc = 0; // not an error - } else if (errno == EAGAIN) { - // delay for a little while if it couldn't get everything out... - uint64_t time_remaining = end_time - get_absolute_time_in_ns(); - useconds_t t = time_remaining / 1000; // ns to us - if (t > 50000) - t = 50000; // wait for 50 milliseconds at most - if (t != 0) { - rc = 0; // not an error - usleep(t); - } - } - } - if (rc > 0) - return count - bytes_remaining; // this is just to mimic a normal write/3. - else - return rc; -} +int try_to_open_pipe_for_writing(const char* pathname) { + // tries to open the pipe in non-blocking mode first. + // if it succeeds, it sets it to blocking. + // if not, it returns -1. -/* -ssize_t non_blocking_write_with_timeout(int fd, const void *buf, size_t count, int timeout) { - // - // timeout is in milliseconds - void *ibuf = (void *)buf; - size_t bytes_remaining = count; - int rc = 1; - struct pollfd ufds[1]; - while ((bytes_remaining > 0) && (rc > 0)) { - // check that we can do some writing - ufds[0].fd = fd; - ufds[0].events = POLLOUT; - rc = poll(ufds, 1, timeout); - if (rc < 0) { - // debug(1, "non-blocking write error waiting for pipe to become ready for writing..."); - } else if (rc == 0) { - // warn("non-blocking write timeout waiting for pipe to become ready for writing"); - rc = -1; - errno = -ETIMEDOUT; - } else { // rc > 0, implying it might be ready - ssize_t bytes_written = write(fd, ibuf, bytes_remaining); - if (bytes_written == -1) { - // debug(1,"Error %d in non_blocking_write: \"%s\".",errno,strerror(errno)); - rc = bytes_written; // to imitate the return from write() - } else { - ibuf += bytes_written; - bytes_remaining -= bytes_written; - } - } - } - if (rc > 0) - return count - bytes_remaining; // this is just to mimic a normal write/3. - else - return rc; - // return write(fd,buf,count); -} -*/ + char errorstring[1024]; + int fdis = open(pathname, O_WRONLY | O_NONBLOCK); // open it in non blocking mode first + // we check that it's not a "real" error. From the "man 2 open" page: + // "ENXIO O_NONBLOCK | O_WRONLY is set, the named file is a FIFO, and no process has the FIFO + // open for reading." Which is okay. -ssize_t non_blocking_write(int fd, const void *buf, size_t count) { - // if it would block, wait up to 0.5 seconds for it to unblock. - return non_blocking_write_with_timeout(fd, buf, count, 500000000); + if ((fdis == -1) && (errno != ENXIO)) { + strerror_r(errno, (char *)errorstring, sizeof(errorstring)); + debug(1, "try_to_open_pipe -- error %d (\"%s\") opening pipe: \"%s\".", errno, + (char *)errorstring, pathname); + } else + if (fdis >= 0) { + // now we switch to blocking mode + int flags = fcntl(fdis, F_GETFL); + if ((flags == -1)) { + strerror_r(errno, (char *)errorstring, sizeof(errorstring)); + debug(1, "try_to_open_pipe -- error %d (\"%s\") getting flags of pipe: \"%s\".", errno, + (char *)errorstring, pathname); + } else { + flags = fcntl(fdis, F_SETFL,flags & ~O_NONBLOCK); + if (flags == -1) { + strerror_r(errno, (char *)errorstring, sizeof(errorstring)); + debug(1, "try_to_open_pipe -- error %d (\"%s\") unsetting NONBLOCK of pipe: \"%s\".", errno, + (char *)errorstring, pathname); + } + } + } + return fdis; } /* from diff --git a/common.h b/common.h index e07cb7c9..b964e051 100644 --- a/common.h +++ b/common.h @@ -303,7 +303,7 @@ int get_requested_connection_state_to_output(); void set_requested_connection_state_to_output(int v); -ssize_t non_blocking_write(int fd, const void *buf, size_t count); // used in a few places +int try_to_open_pipe_for_writing(const char* pathname); // open it without blocking if it's not hooked up /* from * http://coding.debuntu.org/c-implementing-str_replace-replace-all-occurrences-substring#comment-722 diff --git a/rtsp.c b/rtsp.c index a1252133..0a9c0859 100644 --- a/rtsp.c +++ b/rtsp.c @@ -213,6 +213,7 @@ int pc_queue_add_item(pc_queue *the_queue, const void *the_stuff, int block) { i = 0; the_queue->eoq = i; the_queue->count++; + debug(2,"metadata queue+ %d/%d.", the_queue->count, the_queue->capacity); if (the_queue->count == the_queue->capacity) debug(1, "pc_queue is full with %d items in it!", the_queue->count); rc = pthread_cond_signal(&the_queue->pc_queue_item_added_signal); @@ -249,6 +250,7 @@ int pc_queue_get_item(pc_queue *the_queue, void *the_stuff) { i = 0; the_queue->toq = i; the_queue->count--; + debug(2,"metadata queue- %d/%d.", the_queue->count, the_queue->capacity); rc = pthread_cond_signal(&the_queue->pc_queue_item_removed_signal); if (rc) debug(1, "Error signalling after pc_queue_removed_item"); @@ -286,17 +288,15 @@ void *player_watchdog_thread_code(void *arg) { uint64_t last_watchdog_bark_time = conn->watchdog_bark_time; debug_mutex_unlock(&conn->watchdog_mutex, 0); if (last_watchdog_bark_time != 0) { - uint64_t time_since_last_bark = - (get_absolute_time_in_ns() - last_watchdog_bark_time) / 1000000000; + uint64_t time_since_last_bark = (get_absolute_time_in_ns() - last_watchdog_bark_time) / 1000000000; uint64_t ct = config.timeout; // go from int to 64-bit int if (time_since_last_bark >= ct) { conn->watchdog_barks++; if (conn->watchdog_barks == 1) { // debuglev = 3; // tell us everything. - debug(1, - "Connection %d: As Yeats almost said, \"Too long a silence / can make a stone " - "of the heart\".", + debug(1, "Connection %d: As Yeats almost said, \"Too long a silence / can make a stone " + "of the heart\".", conn->connection_number); conn->stop = 1; pthread_cancel(conn->thread); @@ -573,7 +573,7 @@ int msg_handle_line(rtsp_message **pmsg, char *line) { } fail: - debug(3, "msg_handle_line fail"); + debug(3,"msg_handle_line fail"); msg_free(pmsg); *pmsg = NULL; return 0; @@ -616,8 +616,7 @@ enum rtsp_read_request_response rtsp_read_request(rtsp_conn_info *conn, rtsp_mes if (errno == EINTR) continue; if (errno == EAGAIN) { - debug(1, "Connection %d: getting Error 11 -- EAGAIN from a blocking read!", - conn->connection_number); + debug(1, "Connection %d: getting Error 11 -- EAGAIN from a blocking read!", conn->connection_number); continue; } if (errno != ECONNRESET) { @@ -630,15 +629,15 @@ enum rtsp_read_request_response rtsp_read_request(rtsp_conn_info *conn, rtsp_mes goto shutdown; } - /* // this outputs the message received - { - void *pt = malloc(nread+1); - memset(pt, 0, nread+1); - memcpy(pt, buf + inbuf, nread); - debug(1, "Incoming string on port: \"%s\"",pt); - free(pt); - } - */ +/* // this outputs the message received + { + void *pt = malloc(nread+1); + memset(pt, 0, nread+1); + memcpy(pt, buf + inbuf, nread); + debug(1, "Incoming string on port: \"%s\"",pt); + free(pt); + } +*/ inbuf += nread; @@ -647,8 +646,7 @@ enum rtsp_read_request_response rtsp_read_request(rtsp_conn_info *conn, rtsp_mes msg_size = msg_handle_line(the_packet, buf); if (!(*the_packet)) { - debug(1, "Connection %d: rtsp_read_request can't find an RTSP header.", - conn->connection_number); + debug(1,"Connection %d: rtsp_read_request can't find an RTSP header.", conn->connection_number); reply = rtsp_read_request_response_bad_packet; goto shutdown; } @@ -855,10 +853,9 @@ void handle_options(rtsp_conn_info *conn, __attribute__((unused)) rtsp_message * rtsp_message *resp) { debug(3, "Connection %d: OPTIONS", conn->connection_number); resp->respcode = 200; - msg_add_header(resp, "Public", - "ANNOUNCE, SETUP, RECORD, " - "PAUSE, FLUSH, TEARDOWN, " - "OPTIONS, GET_PARAMETER, SET_PARAMETER"); + msg_add_header(resp, "Public", "ANNOUNCE, SETUP, RECORD, " + "PAUSE, FLUSH, TEARDOWN, " + "OPTIONS, GET_PARAMETER, SET_PARAMETER"); } void handle_teardown(rtsp_conn_info *conn, __attribute__((unused)) rtsp_message *req, @@ -999,9 +996,8 @@ void handle_setup(rtsp_conn_info *conn, rtsp_message *req, rtsp_message *resp) { msg_add_header(resp, "Session", "1"); resp->respcode = 200; // it all worked out okay - debug(1, - "Connection %d: SETUP DACP-ID \"%s\" from %s to %s with UDP ports Control: " - "%d, Timing: %d and Audio: %d.", + debug(1, "Connection %d: SETUP DACP-ID \"%s\" from %s to %s with UDP ports Control: " + "%d, Timing: %d and Audio: %d.", conn->connection_number, conn->dacp_id, &conn->client_ip_string, &conn->self_ip_string, conn->local_control_port, conn->local_timing_port, conn->local_audio_port); @@ -1291,11 +1287,7 @@ void metadata_open(void) { char *path = malloc(pl + 1); snprintf(path, pl + 1, "%s", config.metadata_pipename); - fd = open(path, O_WRONLY | O_NONBLOCK); - // if (fd < 0) - // debug(1, "Could not open metadata FIFO %s. Will try again later.", - // path); - + fd = try_to_open_pipe_for_writing(path); free(path); } @@ -1307,8 +1299,8 @@ static void metadata_close(void) { } void metadata_process(uint32_t type, uint32_t code, char *data, uint32_t length) { - // debug(2, "Process metadata with type %x, code %x and length %u.", type, code, length); - int ret; + // debug(1, "Process metadata with type %x, code %x and length %u.", type, code, length); + int ret = 0; if (metadata_sock >= 0 && length < config.metadata_sockmsglength - 8) { char *ptr = metadata_sockmsg; @@ -1366,6 +1358,7 @@ void metadata_process(uint32_t type, uint32_t code, char *data, uint32_t length) } // readers may go away and come back + if (fd < 0) metadata_open(); if (fd < 0) @@ -1373,14 +1366,16 @@ void metadata_process(uint32_t type, uint32_t code, char *data, uint32_t length) char thestring[1024]; snprintf(thestring, 1024, "%x%x%u", type, code, length); - ret = non_blocking_write(fd, thestring, strlen(thestring)); + // ret = non_blocking_write(fd, thestring, strlen(thestring)); + ret = write(fd, thestring, strlen(thestring)); if (ret < 0) { // debug(1,"metadata_process error %d exit 1",ret); return; } if ((data != NULL) && (length > 0)) { snprintf(thestring, 1024, "\n\n"); - ret = non_blocking_write(fd, thestring, strlen(thestring)); + // ret = non_blocking_write(fd, thestring, strlen(thestring)); + ret = write(fd, thestring, strlen(thestring)); if (ret < 0) { // debug(1,"metadata_process error %d exit 2",ret); return; @@ -1403,7 +1398,8 @@ void metadata_process(uint32_t type, uint32_t code, char *data, uint32_t length) debug(1, "Error encoding base64 data."); // debug(1,"Remaining count: %d ret: %d, outbuf_size: // %d.",remaining_count,ret,outbuf_size); - ret = non_blocking_write(fd, outbuf, outbuf_size); + //ret = non_blocking_write(fd, outbuf, outbuf_size); + ret = write(fd, outbuf, outbuf_size); if (ret < 0) { // debug(1,"metadata_process error %d exit 3",ret); return; @@ -1412,14 +1408,16 @@ void metadata_process(uint32_t type, uint32_t code, char *data, uint32_t length) remaining_count -= towrite_count; } snprintf(thestring, 1024, ""); - ret = non_blocking_write(fd, thestring, strlen(thestring)); + // ret = non_blocking_write(fd, thestring, strlen(thestring)); + ret = write(fd, thestring, strlen(thestring)); if (ret < 0) { // debug(1,"metadata_process error %d exit 4",ret); return; } } snprintf(thestring, 1024, "\n"); - ret = non_blocking_write(fd, thestring, strlen(thestring)); + //ret = non_blocking_write(fd, thestring, strlen(thestring)); + ret = write(fd, thestring, strlen(thestring)); if (ret < 0) { // debug(1,"metadata_process error %d exit 5",ret); return; @@ -1451,6 +1449,7 @@ void *metadata_thread_function(__attribute__((unused)) void *ignore) { pthread_cleanup_push(metadata_thread_cleanup_function, NULL); while (1) { pc_queue_get_item(&metadata_queue, &pack); + // debug(1,"pc_queue get item."); pthread_cleanup_push(metadata_pack_cleanup_function, (void *)&pack); if (config.metadata_enabled) { metadata_process(pack.type, pack.code, pack.data, pack.length); @@ -1596,7 +1595,7 @@ static void handle_set_parameter(rtsp_conn_info *conn, rtsp_message *req, rtsp_m char *ct = msg_get_header(req, "Content-Type"); if (ct) { - // debug(2, "SET_PARAMETER Content-Type:\"%s\".", ct); +// debug(2, "SET_PARAMETER Content-Type:\"%s\".", ct); #ifdef CONFIG_METADATA // It seems that the rtptime of the message is used as a kind of an ID that @@ -1894,7 +1893,7 @@ static void handle_announce(rtsp_conn_info *conn, rtsp_message *req, rtsp_messag unsigned int i = 0; unsigned int max_param = sizeof(conn->stream.fmtp) / sizeof(conn->stream.fmtp[0]); - char *found; + char* found; while ((found = strsep(&pfmtp, " \t")) != NULL && i < max_param) { conn->stream.fmtp[i++] = atoi(found); } @@ -2341,7 +2340,6 @@ static void *rtsp_conversation_thread_func(void *pconn) { pthread_condattr_init(&attr); pthread_condattr_setclock(&attr, CLOCK_MONOTONIC); // can't do this in OS X, and don't need it. rc = pthread_cond_init(&conn->flowcontrol, &attr); - pthread_condattr_destroy(&attr); #endif #ifdef COMPILE_FOR_OSX rc = pthread_cond_init(&conn->flowcontrol, NULL); @@ -2376,9 +2374,8 @@ static void *rtsp_conversation_thread_func(void *pconn) { if (strcmp(req->method, "OPTIONS") != 0) // the options message is very common, so don't log it until level 3 debug_level = 2; - debug(debug_level, - "Connection %d: Received an RTSP Packet of type \"%s\":", conn->connection_number, - req->method), + debug(debug_level, "Connection %d: Received an RTSP Packet of type \"%s\":", + conn->connection_number, req->method), debug_print_msg_headers(debug_level, req); apple_challenge(conn->fd, req, resp); @@ -2427,9 +2424,8 @@ static void *rtsp_conversation_thread_func(void *pconn) { if (conn->stop == 0) { int err = msg_write_response(conn->fd, resp); if (err) { - debug(1, - "Connection %d: Unable to write an RTSP message response. Terminating the " - "connection.", + debug(1, "Connection %d: Unable to write an RTSP message response. Terminating the " + "connection.", conn->connection_number); struct linger so_linger; so_linger.l_onoff = 1; // "true" @@ -2482,11 +2478,10 @@ static void *rtsp_conversation_thread_func(void *pconn) { if (reply == -1) { char errorstring[1024]; strerror_r(errno, (char *)errorstring, sizeof(errorstring)); - debug(1, "rtsp_read_request_response_bad_packet write response error %d: \"%s\".", errno, - (char *)errorstring); + debug(1, "rtsp_read_request_response_bad_packet write response error %d: \"%s\".", errno, (char *)errorstring); } else if (reply != (ssize_t)strlen(response_text)) { - debug(1, "rtsp_read_request_response_bad_packet write %d bytes requested but %d written.", - strlen(response_text), reply); + debug(1, "rtsp_read_request_response_bad_packet write %d bytes requested but %d written.", strlen(response_text), + reply); } } else { debug(1, "Connection %d: rtsp_read_request error %d, packet ignored.", @@ -2699,7 +2694,7 @@ void rtsp_listen_loop(void) { debug(2, "Connection %d: new connection from %s:%u to self at %s:%u.", conn->connection_number, remote_ip4, rport, ip4, tport); } -#ifdef AF_INET6 + #ifdef AF_INET6 if (local_info->SAFAMILY == AF_INET6) { // IPv6: @@ -2716,7 +2711,7 @@ void rtsp_listen_loop(void) { debug(2, "Connection %d: new connection from [%s]:%u to self at [%s]:%u.", conn->connection_number, remote_ip6, rport, ip6, tport); } -#endif + #endif } else { debug(1, "Error figuring out Shairport Sync's own IP number."); @@ -2744,8 +2739,8 @@ void rtsp_listen_loop(void) { pthread_cleanup_pop(1); // should never happen } else { warn("could not establish a service on port %d -- program terminating. Is another instance of " - "Shairport Sync running on this device?", - config.port); + "Shairport Sync running on this device?", + config.port); } // debug(1, "Oops -- fell out of the RTSP select loop"); }