Change order of some include file to facilitiate old Mac OS X builds. Check existence of TCP_KEEPINTVL before trying to use it. Fix a race condition with initial messages to the metadata queues. The race was that the queues and mutexes were definied in a thread, so could be delayed to after when the main process was using them.

Author: mikebrady

rtsp.c

@@ -430,7 +430,8 @@ int pc_queue_add_item(pc_queue *the_queue, const void *the_stuff, int block) {
     } else
       rc = pthread_mutex_lock(&the_queue->pc_queue_lock);
     if (rc)
-      debug(1, "Error locking for pc_queue_add_item");
+      debug(1, "Error %d (\"%s\") locking for pc_queue_add_item. Block is %d.", rc, strerror(rc),
+            block);
     pthread_cleanup_push(pc_queue_cleanup_handler, (void *)the_queue);
     // leave this out if you want this to return if the queue is already full
     // irrespective of the block flag.
@@ -1687,7 +1688,7 @@ void handle_get_info(__attribute((unused)) rtsp_conn_info *conn, rtsp_message *r
               hdr);
       }
     }
-    
+
     // In Stage 1, look for the DACP and Active-Remote
     char *ar = msg_get_header(req, "Active-Remote");
     if (ar) {
@@ -2101,8 +2102,8 @@ void handle_post(rtsp_conn_info *conn, rtsp_message *req, rtsp_message *resp) {
   if (strcmp(req->path, "/feedback") == 0) {
     resp->respcode = 501;
   } else {
-    debug(1, "Connection %d: Airplay 1. Unhandled POST %s Content-Length %d", conn->connection_number,
-          req->path, req->contentlength);
+    debug(1, "Connection %d: Airplay 1. Unhandled POST %s Content-Length %d",
+          conn->connection_number, req->path, req->contentlength);
     debug_log_rtsp_message(2, "POST request", req);
   }
 }
@@ -3085,7 +3086,7 @@ void handle_setup_2(rtsp_conn_info *conn, rtsp_message *req, rtsp_message *resp)
                 // iap->ifa_name);
                 if ((iap->ifa_addr) && (iap->ifa_netmask) && (iap->ifa_flags & IFF_UP) &&
                     ((iap->ifa_flags & IFF_LOOPBACK) == 0) &&
-		    (config.interface == NULL || (strcmp(config.interface, iap->ifa_name) == 0))) {
+                    (config.interface == NULL || (strcmp(config.interface, iap->ifa_name) == 0))) {
                   char buf[INET6_ADDRSTRLEN + 1]; // +1 for a NUL
                   memset(buf, 0, sizeof(buf));
                   if (iap->ifa_addr->sa_family == AF_INET6) {
@@ -4081,13 +4082,9 @@ void metadata_pack_cleanup_function(void *arg) {
 void metadata_thread_cleanup_function(__attribute__((unused)) void *arg) {
   // debug(2, "metadata_thread_cleanup_function called");
   metadata_close();
-  pc_queue_delete(&metadata_queue);
 }
 
 void *metadata_thread_function(__attribute__((unused)) void *ignore) {
-  // create a pc_queue for passing information to a threaded metadata handler
-  pc_queue_init(&metadata_queue, (char *)&metadata_queue_items, sizeof(metadata_package),
-                metadata_queue_size, "pipe");
   metadata_create_multicast_socket();
   metadata_package pack;
   pthread_cleanup_push(metadata_thread_cleanup_function, NULL);
@@ -4113,13 +4110,9 @@ void *metadata_thread_function(__attribute__((unused)) void *ignore) {
 void metadata_multicast_thread_cleanup_function(__attribute__((unused)) void *arg) {
   // debug(2, "metadata_multicast_thread_cleanup_function called");
   metadata_delete_multicast_socket();
-  pc_queue_delete(&metadata_multicast_queue);
 }
 
 void *metadata_multicast_thread_function(__attribute__((unused)) void *ignore) {
-  // create a pc_queue for passing information to a threaded metadata handler
-  pc_queue_init(&metadata_multicast_queue, (char *)&metadata_multicast_queue_items,
-                sizeof(metadata_package), metadata_multicast_queue_size, "multicast");
   metadata_create_multicast_socket();
   metadata_package pack;
   pthread_cleanup_push(metadata_multicast_thread_cleanup_function, NULL);
@@ -4154,13 +4147,9 @@ void metadata_hub_close(void) {}
 void metadata_hub_thread_cleanup_function(__attribute__((unused)) void *arg) {
   // debug(2, "metadata_hub_thread_cleanup_function called");
   metadata_hub_close();
-  pc_queue_delete(&metadata_hub_queue);
 }
 
 void *metadata_hub_thread_function(__attribute__((unused)) void *ignore) {
-  // create a pc_queue for passing information to a threaded metadata handler
-  pc_queue_init(&metadata_hub_queue, (char *)&metadata_hub_queue_items, sizeof(metadata_package),
-                metadata_hub_queue_size, "hub");
   metadata_package pack;
   pthread_cleanup_push(metadata_hub_thread_cleanup_function, NULL);
   while (1) {
@@ -4188,14 +4177,10 @@ void metadata_mqtt_close(void) {}
 void metadata_mqtt_thread_cleanup_function(__attribute__((unused)) void *arg) {
   // debug(2, "metadata_mqtt_thread_cleanup_function called");
   metadata_mqtt_close();
-  pc_queue_delete(&metadata_mqtt_queue);
   // debug(2, "metadata_mqtt_thread_cleanup_function done");
 }
 
 void *metadata_mqtt_thread_function(__attribute__((unused)) void *ignore) {
-  // create a pc_queue for passing information to a threaded metadata handler
-  pc_queue_init(&metadata_mqtt_queue, (char *)&metadata_mqtt_queue_items, sizeof(metadata_package),
-                metadata_mqtt_queue_size, "mqtt");
   metadata_package pack;
   pthread_cleanup_push(metadata_mqtt_thread_cleanup_function, NULL);
   while (1) {
@@ -4242,24 +4227,39 @@ void metadata_init(void) {
     if ((fd == -1) && (errno != ENXIO)) {
       char errorstring[1024];
       strerror_r(errno, (char *)errorstring, sizeof(errorstring));
-      debug(1, "metadata_hub_thread_function -- error %d (\"%s\") opening pipe: \"%s\".", errno,
+      debug(1, "metadata_init -- error %d (\"%s\") opening pipe: \"%s\".", errno,
             (char *)errorstring, path);
       warn("can not open metadata pipe -- error %d (\"%s\") opening pipe: \"%s\".", errno,
            (char *)errorstring, path);
     }
     free(path);
+
+    // initialise the metadata queues first, otherwise the might be a race condition
+    // create a pc_queue for the metadata pipe
+    pc_queue_init(&metadata_queue, (char *)&metadata_queue_items, sizeof(metadata_package),
+                  metadata_queue_size, "pipe");
+
     if (pthread_create(&metadata_thread, NULL, metadata_thread_function, NULL) != 0)
       debug(1, "Failed to create metadata thread!");
 
+    // create a pc_queue for the metadata_multicast_queue
+    pc_queue_init(&metadata_multicast_queue, (char *)&metadata_multicast_queue_items,
+                  sizeof(metadata_package), metadata_multicast_queue_size, "multicast");
     if (pthread_create(&metadata_multicast_thread, NULL, metadata_multicast_thread_function,
                        NULL) != 0)
       debug(1, "Failed to create metadata multicast thread!");
   }
 #ifdef CONFIG_METADATA_HUB
+  // create a pc_queue for the metadata hub
+  pc_queue_init(&metadata_hub_queue, (char *)&metadata_hub_queue_items, sizeof(metadata_package),
+                metadata_hub_queue_size, "hub");
   if (pthread_create(&metadata_hub_thread, NULL, metadata_hub_thread_function, NULL) != 0)
     debug(1, "Failed to create metadata hub thread!");
 #endif
 #ifdef CONFIG_MQTT
+  // create a pc_queue for the MQTT handler
+  pc_queue_init(&metadata_mqtt_queue, (char *)&metadata_mqtt_queue_items, sizeof(metadata_package),
+                metadata_mqtt_queue_size, "mqtt");
   if (pthread_create(&metadata_mqtt_thread, NULL, metadata_mqtt_thread_function, NULL) != 0)
     debug(1, "Failed to create metadata mqtt thread!");
 #endif
@@ -4273,26 +4273,30 @@ void metadata_stop(void) {
     // debug(2, "metadata stop mqtt thread.");
     pthread_cancel(metadata_mqtt_thread);
     pthread_join(metadata_mqtt_thread, NULL);
+    pc_queue_delete(&metadata_mqtt_queue);
     // debug(2, "metadata stop mqtt done.");
 #endif
 #ifdef CONFIG_METADATA_HUB
     // debug(2, "metadata stop hub thread.");
     pthread_cancel(metadata_hub_thread);
     pthread_join(metadata_hub_thread, NULL);
     // debug(2, "metadata stop hub done.");
+    pc_queue_delete(&metadata_hub_queue);
 #endif
     if (config.metadata_enabled) {
       // debug(2, "metadata stop multicast thread.");
       if (metadata_multicast_thread) {
         pthread_cancel(metadata_multicast_thread);
         pthread_join(metadata_multicast_thread, NULL);
         // debug(2, "metadata stop multicast done.");
+        pc_queue_delete(&metadata_multicast_queue);
       }
       if (metadata_thread) {
         // debug(2, "metadata stop metadata_thread thread.");
         pthread_cancel(metadata_thread);
         pthread_join(metadata_thread, NULL);
         // debug(2, "metadata_stop finished successfully.");
+        pc_queue_delete(&metadata_queue);
       }
     }
   }
@@ -4343,6 +4347,8 @@ int send_metadata_to_queue(pc_queue *queue, uint32_t type, uint32_t code, char *
     if (data)
       pack.data = memdup(data, length); // only if it's not a null
   }
+
+  // debug(1, "send_metadata_to_queue %x/%x", type, code);
   int rc = pc_queue_add_item(queue, &pack, block);
   if (rc != 0) {
     if (pack.carrier) {
@@ -5642,22 +5648,33 @@ void *rtsp_listen_loop(__attribute((unused)) void *arg) {
       } else {
         size_of_reply = sizeof(SOCKADDR);
         if (getsockname(conn->fd, (struct sockaddr *)&conn->local, &size_of_reply) == 0) {
+
+// skip this stuff in OpenBSD
 #ifndef COMPILE_FOR_OPENBSD
           // Thanks to https://holmeshe.me/network-essentials-setsockopt-SO_KEEPALIVE/ for this.
 
           // turn on keepalive stuff -- wait for keepidle + (keepcnt * keepinttvl time) seconds
           // before giving up an ETIMEOUT error is returned if the keepalive check fails
 
+          // if TCP_KEEPINTVL is defined, check a few times before declaring the line dead
+          // otherwise just wait a little while longer
+
+#ifdef TCP_KEEPINTVL
           int keepAliveIdleTime = 35; // wait this many seconds before checking for a dropped client
           int keepAliveCount = 5;     // check this many times
           int keepAliveInterval = 5;  // wait this many seconds between checks
+#else
+          int keepAliveIdleTime = 60; // wait this many seconds before dropping a client
+#endif
 
+// --- the following is a bit  too complicated
+// decide to use IPPROTO_TCP or SOL_TCP
 #if defined COMPILE_FOR_BSD || defined COMPILE_FOR_OSX
 #define SOL_OPTION IPPROTO_TCP
 #else
 #define SOL_OPTION SOL_TCP
 #endif
-
+// decide to use TCP_KEEPALIVE or TCP_KEEPIDLE
 #ifdef COMPILE_FOR_OSX
 #define KEEP_ALIVE_OR_IDLE_OPTION TCP_KEEPALIVE
 #else
@@ -5666,17 +5683,20 @@ void *rtsp_listen_loop(__attribute((unused)) void *arg) {
 
           if (setsockopt(conn->fd, SOL_OPTION, KEEP_ALIVE_OR_IDLE_OPTION,
                          (void *)&keepAliveIdleTime, sizeof(keepAliveIdleTime))) {
-            debug(1, "can't set the keepidle wait time");
+            debug(1, "can't set the keepAliveIdleTime wait time");
           }
-
+// ---
+// if TCP_KEEPINTVL is defined...
+#ifdef TCP_KEEPINTVL
           if (setsockopt(conn->fd, SOL_OPTION, TCP_KEEPCNT, (void *)&keepAliveCount,
                          sizeof(keepAliveCount))) {
-            debug(1, "can't set the keepidle missing count");
+            debug(1, "can't set the keepAliveCount count");
           }
           if (setsockopt(conn->fd, SOL_OPTION, TCP_KEEPINTVL, (void *)&keepAliveInterval,
                          sizeof(keepAliveInterval))) {
-            debug(1, "can't set the keepidle missing count interval");
+            debug(1, "can't set the keepAliveCount count interval");
           };
+#endif
 #endif
 
           // initialise the connection info
@@ -5736,8 +5756,8 @@ void *rtsp_listen_loop(__attribute((unused)) void *arg) {
     pthread_cleanup_pop(1); // should never happen
   } else {
     die("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");
   pthread_exit(NULL);

shairport.c

@@ -1207,8 +1207,8 @@ int parse_options(int argc, char **argv) {
 
     } else {
       if (config_error_type(&config_file_stuff) == CONFIG_ERR_FILE_IO)
-        die("Error reading configuration file \"%s\": \"%s\".",
-              config_file_real_path, config_error_text(&config_file_stuff));
+        die("Error reading configuration file \"%s\": \"%s\".", config_file_real_path,
+            config_error_text(&config_file_stuff));
       else {
         die("Line %d of the configuration file \"%s\":\n%s", config_error_line(&config_file_stuff),
             config_error_file(&config_file_stuff), config_error_text(&config_file_stuff));
@@ -2110,12 +2110,9 @@ int main(int argc, char **argv) {
 
 #ifdef COMPILE_FOR_OPENBSD
   /* Any command to be executed at runtime? */
-  int run_cmds =
-    config.cmd_active_start != NULL ||
-    config.cmd_active_stop != NULL ||
-    config.cmd_set_volume != NULL ||
-    config.cmd_start != NULL ||
-    config.cmd_stop != NULL;
+  int run_cmds = config.cmd_active_start != NULL || config.cmd_active_stop != NULL ||
+                 config.cmd_set_volume != NULL || config.cmd_start != NULL ||
+                 config.cmd_stop != NULL;
 #endif
 
   // mDNS supports maximum of 63-character names (we append 13).
@@ -2381,11 +2378,11 @@ int main(int argc, char **argv) {
 #ifdef COMPILE_FOR_OPENBSD
   /* Past first and last sio_open(3), sndio(7) only needs "audio". */
 
-# ifdef CONFIG_METADATA
+#ifdef CONFIG_METADATA
   /* Only coverart cache is created.
    * Only metadata pipe is special. */
   if (!config.metadata_enabled)
-# endif
+#endif
   {
     /* Drop "cpath dpath". */
     if (run_cmds) {