opr: Assert opr_cv_timedwait return codes
[openafs.git] / src / fsprobe / fsprobe.c
index a2d9a06..7690992 100644 (file)
@@ -1,7 +1,7 @@
 /*
  * Copyright 2000, International Business Machines Corporation and others.
  * All Rights Reserved.
- * 
+ *
  * This software has been released under the terms of the IBM Public
  * License.  For details, see the LICENSE file in the top-level source
  * directory or online at http://www.openafs.org/dl/license10.html
 #include <afsconfig.h>
 #include <afs/param.h>
 
+#include <roken.h>
 
-
-#include <string.h>
-
-#include "fsprobe.h"           /*Interface for this module */
-#include <lwp.h>               /*Lightweight process package */
+#include <pthread.h>
 #include <afs/cellconfig.h>
 #include <afs/afsint.h>
 #include <afs/afsutil.h>
 #include <afs/volser.h>
 #include <afs/volser_prototypes.h>
-
 #define FSINT_COMMON_XG
 #include <afs/afscbint.h>
 
-#define LWP_STACK_SIZE (16 * 1024)
+#include "fsprobe.h"           /*Interface for this module */
 
 /*
  * Exported variables.
@@ -47,17 +43,11 @@ int fsprobe_ProbeFreqInSecs;        /*Probe freq. in seconds */
 static int fsprobe_initflag = 0;       /*Was init routine called? */
 static int fsprobe_debug = 0;  /*Debugging output enabled? */
 static int (*fsprobe_Handler) (void);  /*Probe handler routine */
-static PROCESS probeLWP_ID;    /*Probe LWP process ID */
+static pthread_t fsprobe_thread;       /*Probe thread */
 static int fsprobe_statsBytes; /*Num bytes in stats block */
 static int fsprobe_probeOKBytes;       /*Num bytes in probeOK block */
-
-/*
- * We have to pass a port to Rx to start up our callback listener
- * service, but 7001 is already taken up by the Cache Manager.  So,
- * we make up our own.
- */
-#define FSPROBE_CBPORT 7101
-
+static opr_mutex_t fsprobe_force_lock; /*Lock to force probe */
+static opr_cv_t fsprobe_force_cv;      /*Condvar to force probe */
 
 /*------------------------------------------------------------------------
  * [private] fsprobe_CleanupInit
@@ -204,7 +194,7 @@ fsprobe_Cleanup(int a_releaseMem)
  * [private] fsprobe_LWP
  *
  * Description:
- *     This LWP iterates over the server connections and gathers up
+ *     This thread iterates over the server connections and gathers up
  *     the desired statistics from each one on a regular basis.  When
  *     the sweep is done, the associated handler function is called
  *     to process the new data.
@@ -226,15 +216,16 @@ fsprobe_LWP(void *unused)
 {                              /*fsprobe_LWP */
 
     static char rn[] = "fsprobe_LWP";  /*Routine name */
-    register afs_int32 code;   /*Results of calls */
+    afs_int32 code;    /*Results of calls */
     struct timeval tv;         /*Time structure */
+    struct timespec wait;      /*Time to wait */
     int conn_idx;              /*Connection index */
     struct fsprobe_ConnectionInfo *curr_conn;  /*Current connection */
     struct ProbeViceStatistics *curr_stats;    /*Current stats region */
     int *curr_probeOK;         /*Current probeOK field */
     ViceStatistics64 stats64;      /*Current stats region */
-    stats64.ViceStatistics64_val = (afs_uint64 *)malloc(STATS64_VERSION *
-                                                       sizeof(afs_uint64));
+    stats64.ViceStatistics64_val = malloc(STATS64_VERSION *
+                                         sizeof(afs_uint64));
     while (1) {                        /*Service loop */
        /*
         * Iterate through the server connections, gathering data.
@@ -293,7 +284,7 @@ fsprobe_LWP(void *unused)
                char pname[10];
                struct diskPartition partition;
                struct diskPartition64 *partition64p =
-                   (struct diskPartition64 *)malloc(sizeof(struct diskPartition64));
+                   malloc(sizeof(struct diskPartition64));
 
                if (fsprobe_debug)
                    fprintf(stderr,
@@ -315,8 +306,8 @@ fsprobe_LWP(void *unused)
                            strcpy(curr_stats->Disk[i].Name, pname);
                        }
                        if (code == RXGEN_OPCODE) {
-                           code = 
-                               AFSVolPartitionInfo(curr_conn->rxVolconn, 
+                           code =
+                               AFSVolPartitionInfo(curr_conn->rxVolconn,
                                                    pname, &partition);
                            if (!code) {
                                curr_stats->Disk[i].BlocksAvailable =
@@ -362,24 +353,18 @@ fsprobe_LWP(void *unused)
                    rn, code);
 
        /*
-        * Fall asleep for the prescribed number of seconds.
+        * Fall asleep for the prescribed number of seconds or wakeup
+        * sooner if forced.
         */
-       tv.tv_sec = fsprobe_ProbeFreqInSecs;
-       tv.tv_usec = 0;
-       if (fsprobe_debug)
-           fprintf(stderr, "[%s] Falling asleep for %d seconds\n", rn,
-                   fsprobe_ProbeFreqInSecs);
-       code = IOMGR_Select(0,  /*Num fids */
-                           0,  /*Descriptors ready for reading */
-                           0,  /*Descriptors ready for writing */
-                           0,  /*Descriptors w/exceptional conditions */
-                           &tv);       /*Ptr to timeout structure */
-       if (code)
-           fprintf(stderr, "[%s] IOMGR_Select returned code %d\n", rn, code);
+       gettimeofday(&tv, NULL);
+       wait.tv_sec = tv.tv_sec + fsprobe_ProbeFreqInSecs;
+       wait.tv_nsec = tv.tv_usec * 1000;
+       opr_mutex_enter(&fsprobe_force_lock);
+       code = opr_cv_timedwait(&fsprobe_force_cv, &fsprobe_force_lock, &wait);
+       opr_mutex_exit(&fsprobe_force_lock);
     }                          /*Service loop */
-    free(stats64.ViceStatistics64_val);
-
-    return NULL;
+    AFS_UNREACHED(free(stats64.ViceStatistics64_val));
+    AFS_UNREACHED(return(NULL));
 }                              /*fsprobe_LWP */
 
 /*list all the partitions on <aserver> */
@@ -391,7 +376,7 @@ XListPartitions(struct rx_connection *aconn, struct partList *ptrPartList,
 {
     struct pIDs partIds;
     struct partEntries partEnts;
-    register int i, j = 0, code;
+    int i, j = 0, code;
 
     *cntp = 0;
     if (newvolserver == 1) {
@@ -450,7 +435,7 @@ XListPartitions(struct rx_connection *aconn, struct partList *ptrPartList,
  *
  * Description:
  *     Initialize the fsprobe module: set up Rx connections to the
- *     given set of servers, start up the probe and callback LWPs,
+ *     given set of servers, start up the probe and callback threads,
  *     and associate the routine to be called when a probe completes.
  *
  * Arguments:
@@ -463,7 +448,7 @@ XListPartitions(struct rx_connection *aconn, struct partList *ptrPartList,
  * Returns:
  *     0 on success,
  *     -2 for (at least one) connection error,
- *     LWP process creation code, if it failed,
+ *     thread process creation code, if it failed,
  *     -1 for other fatal errors.
  *
  * Environment:
@@ -476,13 +461,13 @@ XListPartitions(struct rx_connection *aconn, struct partList *ptrPartList,
  *------------------------------------------------------------------------*/
 
 int
-fsprobe_Init(int a_numServers, struct sockaddr_in *a_socketArray, 
+fsprobe_Init(int a_numServers, struct sockaddr_in *a_socketArray,
             int a_ProbeFreqInSecs, int (*a_ProbeHandler)(void),
             int a_debug)
 {                              /*fsprobe_Init */
 
     static char rn[] = "fsprobe_Init"; /*Routine name */
-    register afs_int32 code;   /*Return value */
+    afs_int32 code;    /*Return value */
     static struct rx_securityClass *CBsecobj;  /*Callback security object */
     struct rx_securityClass *secobj;   /*Client security object */
     struct rx_service *rxsrv_afsserver;        /*Server for AFS */
@@ -491,7 +476,6 @@ fsprobe_Init(int a_numServers, struct sockaddr_in *a_socketArray,
     struct fsprobe_ConnectionInfo *curr_conn;  /*Ptr to current conn */
     char *hostNameFound;       /*Ptr to returned host name */
     int conn_err;              /*Connection error? */
-    int PortToUse;             /*Callback port to use */
 
     /*
      * If we've already been called, snicker at the bozo, gently
@@ -503,6 +487,9 @@ fsprobe_Init(int a_numServers, struct sockaddr_in *a_socketArray,
     } else
        fsprobe_initflag = 1;
 
+    opr_mutex_init(&fsprobe_force_lock);
+    opr_cv_init(&fsprobe_force_cv);
+
     /*
      * Check the parameters for bogosities.
      */
@@ -521,7 +508,7 @@ fsprobe_Init(int a_numServers, struct sockaddr_in *a_socketArray,
                a_ProbeFreqInSecs);
        arg_errfound = 1;
     }
-    if (a_ProbeHandler == (int (*)())0) {
+    if (a_ProbeHandler == NULL) {
        fprintf(stderr, "[%s] Null probe handler function argument\n", rn);
        arg_errfound = 1;
     }
@@ -555,11 +542,6 @@ fsprobe_Init(int a_numServers, struct sockaddr_in *a_socketArray,
                (a_numServers * sizeof(struct fsprobe_ConnectionInfo)));
        return (-1);            /*No cleanup needs to be done yet */
     }
-#if 0
-    else
-       fprintf(stderr, "[%s] fsprobe_ConnInfo allocated (%d bytes)\n", rn,
-               a_numServers * sizeof(struct fsprobe_ConnectionInfo));
-#endif /* 0 */
 
     fsprobe_statsBytes = a_numServers * sizeof(struct ProbeViceStatistics);
     fsprobe_Results.stats = (struct ProbeViceStatistics *)
@@ -575,7 +557,7 @@ fsprobe_Init(int a_numServers, struct sockaddr_in *a_socketArray,
                rn, fsprobe_statsBytes);
 
     fsprobe_probeOKBytes = a_numServers * sizeof(int);
-    fsprobe_Results.probeOK = (int *)malloc(fsprobe_probeOKBytes);
+    fsprobe_Results.probeOK = malloc(fsprobe_probeOKBytes);
     if (fsprobe_Results.probeOK == (int *)0) {
        fprintf(stderr,
                "[%s] Can't allocate %d probeOK array entries (%d bytes)\n",
@@ -596,24 +578,13 @@ fsprobe_Init(int a_numServers, struct sockaddr_in *a_socketArray,
      */
     if (fsprobe_debug)
        fprintf(stderr, "[%s] Initializing Rx\n", rn);
-    PortToUse = FSPROBE_CBPORT;
-    do {
-       code = rx_Init(htons(PortToUse));
-       if (code) {
-           if (code == RX_ADDRINUSE) {
-               if (fsprobe_debug)
-                   fprintf(stderr,
-                           "[%s] Callback port %d in use, advancing\n", rn,
-                           PortToUse);
-               PortToUse++;
-           } else {
-               fprintf(stderr, "[%s] Fatal error in rx_Init()\n", rn);
-               return (-1);
-           }
-       }
-    } while (code);
+    code = rx_Init(0);
+    if (code) {
+       fprintf(stderr, "[%s] Fatal error in rx_Init()\n", rn);
+       return (-1);
+    }
     if (fsprobe_debug)
-       fprintf(stderr, "[%s] Rx initialized on port %d\n", rn, PortToUse);
+       fprintf(stderr, "[%s] Rx initialized.\n", rn);
 
     /*
      * Create a null Rx server security object, to be used by the
@@ -632,18 +603,18 @@ fsprobe_Init(int a_numServers, struct sockaddr_in *a_socketArray,
 
     /*
      * Create a null Rx client security object, to be used by the
-     * probe LWP.
+     * probe thread.
      */
     secobj = rxnull_NewClientSecurityObject();
     if (secobj == (struct rx_securityClass *)0) {
        fprintf(stderr,
-               "[%s] Can't create client security object for probe LWP.\n",
+               "[%s] Can't create client security object for probe thread.\n",
                rn);
        fsprobe_Cleanup(1);     /*Delete already-malloc'ed areas */
        return (-1);
     }
     if (fsprobe_debug)
-       fprintf(stderr, "[%s] Probe LWP client security object created\n",
+       fprintf(stderr, "[%s] Probe thread client security object created\n",
                rn);
 
     curr_conn = fsprobe_ConnInfo;
@@ -766,37 +737,20 @@ fsprobe_Init(int a_numServers, struct sockaddr_in *a_socketArray,
      */
     if (fsprobe_debug)
        fprintf(stderr, "[%s] Starting up callback listener.\n", rn);
-    rx_StartServer(0 /*Don't donate yourself to LWP pool */ );
+    rx_StartServer(0 /*Don't donate yourself to thread pool */ );
 
     /*
-     * Start up the probe LWP.
+     * Start up the probe thread.
      */
     if (fsprobe_debug)
-       fprintf(stderr, "[%s] Creating the probe LWP\n", rn);
-    code = LWP_CreateProcess(fsprobe_LWP,      /*Function to start up */
-                            LWP_STACK_SIZE,    /*Stack size in bytes */
-                            1, /*Priority */
-                            (void *)0, /*Parameters */
-                            "fsprobe Worker",  /*Name to use */
-                            &probeLWP_ID);     /*Returned LWP process ID */
+       fprintf(stderr, "[%s] Creating the probe thread\n", rn);
+    code = pthread_create(&fsprobe_thread, NULL, fsprobe_LWP, NULL);
     if (code) {
-       fprintf(stderr, "[%s] Can't create fsprobe LWP!  Error is %d\n", rn,
+       fprintf(stderr, "[%s] Can't create fsprobe thread!  Error is %d\n", rn,
                code);
        fsprobe_Cleanup(1);     /*Delete already-malloc'ed areas */
        return (code);
     }
-    if (fsprobe_debug)
-       fprintf(stderr, "[%s] Probe LWP process structure located at %p\n",
-               rn, probeLWP_ID);
-
-#if 0
-    /*
-     * Do I need to do this?
-     */
-    if (fsprobe_debug)
-       fprintf(stderr, "[%s] Calling osi_Wakeup()\n", rn);
-    osi_Wakeup(&rxsrv_afsserver);      /*Wake up anyone waiting for it */
-#endif /* 0 */
 
     /*
      * Return the final results.
@@ -813,7 +767,7 @@ fsprobe_Init(int a_numServers, struct sockaddr_in *a_socketArray,
  * [exported] fsprobe_ForceProbeNow
  *
  * Description:
- *     Wake up the probe LWP, forcing it to execute a probe immediately.
+ *     Wake up the probe thread, forcing it to execute a probe immediately.
  *
  * Arguments:
  *     None.
@@ -846,7 +800,9 @@ fsprobe_ForceProbeNow(void)
     /*
      * Kick the sucker in the side.
      */
-    IOMGR_Cancel(probeLWP_ID);
+    opr_mutex_enter(&fsprobe_force_lock);
+    opr_cv_signal(&fsprobe_force_cv);
+    opr_mutex_exit(&fsprobe_force_lock);
 
     /*
      * We did it, so report the happy news.
@@ -854,3 +810,44 @@ fsprobe_ForceProbeNow(void)
     return (0);
 
 }                              /*fsprobe_ForceProbeNow */
+
+/*------------------------------------------------------------------------
+ * [exported] fsprobe_Wait
+ *
+ * Description:
+ *     Wait for the collection to complete.
+ *
+ * Arguments:
+ *    int sleep_secs : time to wait in seconds. 0 means sleep forever.
+ *
+ * Returns:
+ *     0 on success,
+ *     Error value otherwise.
+ *
+ * Environment:
+ *     The module must have been initialized.
+ *
+ * Side Effects:
+ *     As advertised.
+ *------------------------------------------------------------------------*/
+int
+fsprobe_Wait(int sleep_secs)
+{
+    int code;
+    struct timeval tv;
+
+    if (sleep_secs == 0) {
+       while (1) {
+           tv.tv_sec = 30;
+           tv.tv_usec = 0;
+           code = select(0, 0, 0, 0, &tv);
+           if (code < 0)
+               break;
+       }
+    } else {
+       tv.tv_sec = sleep_secs;
+       tv.tv_usec = 0;
+       code = select(0, 0, 0, 0, &tv);
+    }
+    return code;
+}