Graceful shutdown closes sockets

build/darknet/x64/darknet_demo_json_stream.cmd

@@ -1,7 +1,7 @@
 rem Run this file and then open URL in Chrome/Firefox: rem http://localhost:8070
 rem Or open: http://ip-address:8070
 
-darknet.exe detector demo cfg/coco.data cfg/yolov3.cfg yolov3.weights test.mp4 -i 0 -json_port 8070 -dont_show -ext_output
+darknet.exe detector demo cfg/coco.data cfg/yolov3.cfg yolov3.weights test.mp4 -i 0 -json_port 8070 -ext_output
 
 
 pause

include/darknet.h

@@ -744,7 +744,7 @@ LIB_API detection *get_network_boxes(network *net, int w, int h, float thresh, f
 LIB_API void free_detections(detection *dets, int n);
 LIB_API void fuse_conv_batchnorm(network net);
 LIB_API void calculate_binary_weights(network net);
-LIB_API char *detection_to_json(detection *dets, int nboxes, int classes, char **names, long long int frame_id);
+LIB_API char *detection_to_json(detection *dets, int nboxes, int classes, char **names, long long int frame_id, char *filename);
 
 LIB_API layer* get_network_layer(network* net, int i);
 //LIB_API detection *get_network_boxes(network *net, int w, int h, float thresh, float hier, int *map, int relative, int *num, int letter);

src/http_stream.cpp

@@ -58,13 +58,14 @@ using namespace cv;
 #include "image.h"
 
 
-class MJPGWriter
+class MJPG_sender
 {
     SOCKET sock;
     SOCKET maxfd;
     fd_set master;
     int timeout; // master sock timeout, shutdown after timeout millis.
     int quality; // jpeg compression [1..100]
+    int close_all_sockets;
 
     int _write(int sock, char const*const s, int len)
     {
@@ -74,18 +75,20 @@ class MJPGWriter
 
 public:
 
-    MJPGWriter(int port = 0, int _timeout = 200000, int _quality = 30)
+    MJPG_sender(int port = 0, int _timeout = 200000, int _quality = 30)
         : sock(INVALID_SOCKET)
         , timeout(_timeout)
         , quality(_quality)
     {
+        close_all_sockets = 0;
         FD_ZERO(&master);
         if (port)
             open(port);
     }
 
-    ~MJPGWriter()
+    ~MJPG_sender()
     {
+        close_all();
         release();
     }
 
@@ -97,6 +100,13 @@ public:
         return false;
     }
 
+    void close_all()
+    {
+        close_all_sockets = 1;
+        cv::Mat tmp(cv::Size(10, 10), CV_8UC3);
+        write(tmp);
+    }
+
     bool open(int port)
     {
         sock = ::socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
@@ -107,12 +117,12 @@ public:
         address.sin_port = htons(port);    // ::htons(port);
         if (::bind(sock, (SOCKADDR*)&address, sizeof(SOCKADDR_IN)) == SOCKET_ERROR)
         {
-            cerr << "error : couldn't bind sock " << sock << " to port " << port << "!" << endl;
+            cerr << "error MJPG_sender: couldn't bind sock " << sock << " to port " << port << "!" << endl;
             return release();
         }
         if (::listen(sock, 10) == SOCKET_ERROR)
         {
-            cerr << "error : couldn't listen on sock " << sock << " on port " << port << " !" << endl;
+            cerr << "error MJPG_sender: couldn't listen on sock " << sock << " on port " << port << " !" << endl;
             return release();
         }
         FD_ZERO(&master);
@@ -158,9 +168,15 @@ public:
                 SOCKET      client = ::accept(sock, (SOCKADDR*)&address, &addrlen);
                 if (client == SOCKET_ERROR)
                 {
-                    cerr << "error : couldn't accept connection on sock " << sock << " !" << endl;
+                    cerr << "error MJPG_sender: couldn't accept connection on sock " << sock << " !" << endl;
                     return false;
                 }
+                if (setsockopt(client, SOL_SOCKET, SO_RCVTIMEO, (char *)&to, sizeof(to)) < 0) {
+                    cerr << "error MJPG_sender: SO_RCVTIMEO setsockopt failed\n";
+                }
+                if (setsockopt(client, SOL_SOCKET, SO_SNDTIMEO, (char *)&to, sizeof(to)) < 0) {
+                    cerr << "error MJPG_sender: SO_SNDTIMEO setsockopt failed\n";
+                }
                 maxfd = (maxfd>client ? maxfd : client);
                 FD_SET(client, &master);
                 _write(client, "HTTP/1.0 200 OK\r\n", 0);
@@ -174,10 +190,20 @@ public:
                     "Pragma: no-cache\r\n"
                     "Content-Type: multipart/x-mixed-replace; boundary=mjpegstream\r\n"
                     "\r\n", 0);
-                cerr << "new client " << client << endl;
+                cerr << "MJPG_sender: new client " << client << endl;
             }
             else // existing client, just stream pix
             {
+                if (close_all_sockets) {
+                    int result = ::shutdown(s, 1); // 0 close input, 1 close output, 2 close both
+                    char *buf = (char *)calloc(1024, sizeof(char));
+                    ::recv(s, buf, 1024, 0);
+                    free(buf);
+                    closesocket(s);
+                    printf("MJPG_sender: close clinet: %d \n", result);
+                    continue;
+                }
+
                 char head[400];
                 sprintf(head, "--mjpegstream\r\nContent-Type: image/jpeg\r\nContent-Length: %zu\r\n\r\n", outlen);
                 _write(s, head, 0);
@@ -185,7 +211,7 @@ public:
                 //cerr << "known client " << s << " " << n << endl;
                 if (n < outlen)
                 {
-                    cerr << "kill client " << s << endl;
+                    cerr << "MJPG_sender: kill client " << s << endl;
                     ::shutdown(s, 2);
                     FD_CLR(s, &master);
                 }
@@ -198,7 +224,7 @@ public:
 
 void send_mjpeg(IplImage* ipl, int port, int timeout, int quality)
 {
-    static MJPGWriter wri(port, timeout, quality);
+    static MJPG_sender wri(port, timeout, quality);
     cv::Mat mat = cv::cvarrToMat(ipl);
     wri.write(mat);
     std::cout << " MJPEG-stream sent. \n";
@@ -211,6 +237,7 @@ class JSON_sender
     SOCKET maxfd;
     fd_set master;
     int timeout; // master sock timeout, shutdown after timeout millis.
+    int close_all_sockets;
 
     int _write(int sock, char const*const s, int len)
     {
@@ -224,6 +251,7 @@ public:
         : sock(INVALID_SOCKET)
         , timeout(_timeout)
     {
+        close_all_sockets = 0;
         FD_ZERO(&master);
         if (port)
             open(port);
@@ -231,6 +259,7 @@ public:
 
     ~JSON_sender()
     {
+        close_all();
         release();
     }
 
@@ -242,6 +271,13 @@ public:
         return false;
     }
 
+    void close_all()
+    {
+        close_all_sockets = 1;
+        char tmp[1];
+        write(tmp);
+    }
+
     bool open(int port)
     {
         sock = ::socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
@@ -252,12 +288,12 @@ public:
         address.sin_port = htons(port);    // ::htons(port);
         if (::bind(sock, (SOCKADDR*)&address, sizeof(SOCKADDR_IN)) == SOCKET_ERROR)
         {
-            cerr << "error : couldn't bind sock " << sock << " to port " << port << "!" << endl;
+            cerr << "error JSON_sender: couldn't bind sock " << sock << " to port " << port << "!" << endl;
             return release();
         }
         if (::listen(sock, 10) == SOCKET_ERROR)
         {
-            cerr << "error : couldn't listen on sock " << sock << " on port " << port << " !" << endl;
+            cerr << "error JSON_sender: couldn't listen on sock " << sock << " on port " << port << " !" << endl;
             return release();
         }
         FD_ZERO(&master);
@@ -298,9 +334,15 @@ public:
                 SOCKET      client = ::accept(sock, (SOCKADDR*)&address, &addrlen);
                 if (client == SOCKET_ERROR)
                 {
-                    cerr << "error : couldn't accept connection on sock " << sock << " !" << endl;
+                    cerr << "error JSON_sender: couldn't accept connection on sock " << sock << " !" << endl;
                     return false;
                 }
+                if (setsockopt(client, SOL_SOCKET, SO_RCVTIMEO, (char *)&to, sizeof(to)) < 0) {
+                    cerr << "error JSON_sender: SO_RCVTIMEO setsockopt failed\n";
+                }
+                if (setsockopt(client, SOL_SOCKET, SO_SNDTIMEO, (char *)&to, sizeof(to)) < 0) {
+                    cerr << "error JSON_sender: SO_SNDTIMEO setsockopt failed\n";
+                }
                 maxfd = (maxfd>client ? maxfd : client);
                 FD_SET(client, &master);
                 _write(client, "HTTP/1.0 200 OK\r\n", 0);
@@ -315,23 +357,37 @@ public:
                     "Content-Type: application/json\r\n"
                     //"Content-Type: multipart/x-mixed-replace; boundary=boundary\r\n"
                     "\r\n", 0);
-                cerr << "new client " << client << endl;
+                cerr << "JSON_sender: new client " << client << endl;
             }
             else // existing client, just stream pix
             {
-                char head[400];
+                // Graceful closes will first close their output channels and then wait for the peer
+                // on the other side of the connection to close its output channels. When both sides are done telling
+                // each other they won’t be sending any more data (i.e., closing output channels),
+                // the connection can be closed fully, with no risk of reset.
+                if (close_all_sockets) {
+                    int result = ::shutdown(s, 1); // 0 close input, 1 close output, 2 close both
+                    char *buf = (char *)calloc(1024, sizeof(char));
+                    ::recv(s, buf, 1024, 0);
+                    free(buf);
+                    closesocket(s);
+                    printf("JSON_sender: close clinet: %d \n", result);
+                    continue;
+                }
+
+                //char head[400];
                 // application/json
                 // application/x-resource+json or application/x-collection+json -  when you are representing REST resources and collections
                 // text/json or text/javascript or text/plain.
                 // https://stackoverflow.com/questions/477816/what-is-the-correct-json-content-type
                 //sprintf(head, "\r\nContent-Length: %zu\r\n\r\n", outlen);
                 //sprintf(head, "--boundary\r\nContent-Type: application/json\r\nContent-Length: %zu\r\n\r\n", outlen);
-                _write(s, head, 0);
+                //_write(s, head, 0);
                 int n = _write(s, outputbuf, outlen);
                 //cerr << "known client " << s << " " << n << endl;
                 if (n < outlen)
                 {
-                    cerr << "kill client " << s << endl;
+                    cerr << "JSON_sender: kill client " << s << endl;
                     ::shutdown(s, 2);
                     FD_CLR(s, &master);
                 }
@@ -342,19 +398,10 @@ public:
 };
 // ----------------------------------------
 
-// JSON format:
-//{
-// "frame_id":8990,
-// "objects":[
-//  {"class_id":4, "name":"aeroplane", "relative coordinates":{"center_x":0.398831, "center_y":0.630203, "width":0.057455, "height":0.020396}, "confidence":0.793070},
-//  {"class_id":14, "name":"bird", "relative coordinates":{"center_x":0.398831, "center_y":0.630203, "width":0.057455, "height":0.020396}, "confidence":0.265497}
-// ]
-//},
-
 void send_json(detection *dets, int nboxes, int classes, char **names, long long int frame_id, int port, int timeout)
 {
     static JSON_sender js(port, timeout);
-    char *send_buf = detection_to_json(dets, nboxes, classes, names, frame_id);
+    char *send_buf = detection_to_json(dets, nboxes, classes, names, frame_id, NULL);
 
     js.write(send_buf);
     std::cout << " JSON-stream sent. \n";

src/network.c

@@ -658,12 +658,26 @@ void free_detections(detection *dets, int n)
     free(dets);
 }
 
-char *detection_to_json(detection *dets, int nboxes, int classes, char **names, long long int frame_id)
+// JSON format:
+//{
+// "frame_id":8990,
+// "objects":[
+//  {"class_id":4, "name":"aeroplane", "relative coordinates":{"center_x":0.398831, "center_y":0.630203, "width":0.057455, "height":0.020396}, "confidence":0.793070},
+//  {"class_id":14, "name":"bird", "relative coordinates":{"center_x":0.398831, "center_y":0.630203, "width":0.057455, "height":0.020396}, "confidence":0.265497}
+// ]
+//},
+
+char *detection_to_json(detection *dets, int nboxes, int classes, char **names, long long int frame_id, char *filename)
 {
     const float thresh = 0.005; // function get_network_boxes() has already filtred dets by actual threshold
 
     char *send_buf = (char *)calloc(1024, sizeof(char));
-    sprintf(send_buf, "{\n \"frame_id\":%d, \n \"objects\":[ \n", frame_id);
+    if (filename) {
+        sprintf(send_buf, "{\n \"frame_id\":%d, \n \"filename\":\"%s\", \n \"objects\": [ \n", frame_id, filename);
+    }
+    else {
+        sprintf(send_buf, "{\n \"frame_id\":%d, \n \"objects\": [ \n", frame_id);
+    }
 
     int i, j;
     int class_id = -1;