af-packet: fix if/down issues with tpacket-v2/autofp

The AFPSwitchState function would close the socket and free the
other resources when the interface went down _and_ the ref cnt was
0. However in autofp mode it was common to get to this point while
packets were still processed in the autofp worker threads, meaning
the ref cnt would not be 0. On the interface coming back up the
initialization code would overwrite the socket and rings, leading
to resource leaks.

Socket ref cnt is decremented from the v2 release callback. If the
callback would get to ref cnt 0, the packet would not be released
in the kernel, but it would (possibly) close the socket if the
iface was down, but not free other resources.

This patch changes the logic to first release the packet to the
kernel and then decrement the ref cnt and it makes the main receive
loop the only one responsible for opening and closing sockets. Wait
with closing the socket and rings until the ref count is 0, which can
happen after AFPSwitchState is called due to packets still being
processed by autofp worker threads.

Bug: #4803.

src/source-af-packet.c

@@ -858,16 +858,14 @@ static void AFPReleaseDataFromRing(Packet *p)
         AFPWritePacket(p, TPACKET_V2);
     }
 
-    if (AFPDerefSocket(p->afp_v.mpeer) == 0)
-        goto cleanup;
-
     if (p->afp_v.relptr) {
         union thdr h;
         h.raw = p->afp_v.relptr;
         h.h2->tp_status = TP_STATUS_KERNEL;
     }
 
-cleanup:
+    (void)AFPDerefSocket(p->afp_v.mpeer);
+
     AFPV_CLEANUP(&p->afp_v);
 }
 
@@ -1264,52 +1262,52 @@ static int AFPDerefSocket(AFPPeer* peer)
         return 1;
 
     if (SC_ATOMIC_SUB(peer->sock_usage, 1) == 1) {
-        if (SC_ATOMIC_GET(peer->state) == AFP_STATE_DOWN) {
-            SCLogInfo("Cleaning socket connected to '%s'", peer->iface);
-            close(SC_ATOMIC_GET(peer->socket));
-            return 0;
-        }
+        return 0;
     }
     return 1;
 }
 
-static void AFPSwitchState(AFPThreadVars *ptv, int state)
+static void AFPCloseSocket(AFPThreadVars *ptv)
 {
-    ptv->afp_state = state;
-    ptv->down_count = 0;
-
-    AFPPeerUpdate(ptv);
+    if (ptv->mpeer != NULL)
+        BUG_ON(SC_ATOMIC_GET(ptv->mpeer->sock_usage) != 0);
 
-    /* Do cleaning if switching to down state */
-    if (state == AFP_STATE_DOWN) {
-#ifdef HAVE_TPACKET_V3
-        if (ptv->flags & AFP_TPACKET_V3) {
-            if (!ptv->ring.v3) {
-                SCFree(ptv->ring.v3);
-                ptv->ring.v3 = NULL;
-            }
-        } else {
-#endif
-            if (ptv->ring.v2) {
-                /* only used in reading phase, we can free it */
-                SCFree(ptv->ring.v2);
-                ptv->ring.v2 = NULL;
-            }
+    if (ptv->flags & AFP_TPACKET_V3) {
 #ifdef HAVE_TPACKET_V3
+        if (ptv->ring.v3) {
+            SCFree(ptv->ring.v3);
+            ptv->ring.v3 = NULL;
         }
 #endif
-        if (ptv->socket != -1) {
-            /* we need to wait for all packets to return data */
-            if (SC_ATOMIC_SUB(ptv->mpeer->sock_usage, 1) == 1) {
-                SCLogDebug("Cleaning socket connected to '%s'", ptv->iface);
-                munmap(ptv->ring_buf, ptv->ring_buflen);
-                close(ptv->socket);
-                ptv->socket = -1;
-            }
+    } else {
+        if (ptv->ring.v2) {
+            /* only used in reading phase, we can free it */
+            SCFree(ptv->ring.v2);
+            ptv->ring.v2 = NULL;
         }
     }
+    if (ptv->socket != -1) {
+        SCLogDebug("Cleaning socket connected to '%s'", ptv->iface);
+        munmap(ptv->ring_buf, ptv->ring_buflen);
+        close(ptv->socket);
+        ptv->socket = -1;
+    }
+}
+
+static void AFPSwitchState(AFPThreadVars *ptv, int state)
+{
+    ptv->afp_state = state;
+    ptv->down_count = 0;
+
+    if (state == AFP_STATE_DOWN) {
+        /* cleanup is done on thread cleanup or try reopen
+         * as there may still be packets in autofp that
+         * are referencing us */
+        (void)SC_ATOMIC_SUB(ptv->mpeer->sock_usage, 1);
+    }
     if (state == AFP_STATE_UP) {
-         (void)SC_ATOMIC_SET(ptv->mpeer->sock_usage, 1);
+        AFPPeerUpdate(ptv);
+        (void)SC_ATOMIC_SET(ptv->mpeer->sock_usage, 1);
     }
 }
 
@@ -1480,6 +1478,9 @@ static int AFPTryReopen(AFPThreadVars *ptv)
         return -1;
     }
 
+    /* ref cnt 0, we can close the old socket */
+    AFPCloseSocket(ptv);
+
     int afp_activate_r = AFPCreateSocket(ptv, ptv->iface, 0);
     if (afp_activate_r != 0) {
         if (ptv->down_count % AFP_DOWN_COUNTER_INTERVAL == 0) {