
IMPLEMENTATION MODULE Stats;

        (********************************************************)
        (*                                                      *)
        (*                  Traffic statistics                  *)
        (*                                                      *)
        (*  Programmer:         P. Moylan                       *)
        (*  Last edited:        1 April 2024                    *)
        (*  Status:             Working                         *)
        (*                                                      *)
        (*  We estimate arrival rate and mean service time for  *)
        (*  four kinds of traffic: POP3, SMTP normal, SMTP via  *)
        (*  submission port, and outgoing mail.                 *)
        (*                                                      *)
        (********************************************************)


(************************************************************************)
(*                                                                      *)
(* Still two important decisions to make about time.                    *)
(* 1. Integer or floating point?  My simulations required floating      *)
(*    point because simulation dt had to be much less than a second.    *)
(*    For estimation we'd like better precision but probably can't get  *)
(*    it, but the DATETIME structure does allow for hundredths.         *)
(* 2. How to calculate current time?  DosGetDateTime returns a record   *)
(*    with year/month/day etc, which requires some work to convert to   *)
(*    seconds, but for now I can't see a better way.                    *)
(*                                                                      *)
(* Also to note: we have approx 31555872 seconds/year.  A 32-bit        *)
(* cardinal can hold up to 4.294967E9, which allows for 136 years.  If  *)
(* the system base time was 1970, that takes us well past the probable  *)
(* lifetime of this software, so LONGCARD is not needed.  The Unix 2038 *)
(* problem is because Unix software stores _signed_ seconds.            *)
(*                                                                      *)
(************************************************************************)

IMPORT INIData, OS2;

FROM TimeConv IMPORT
    (* proc *)  time;

FROM WINI IMPORT
    (* proc *)  OpenINI, CloseINI;

FROM INIData IMPORT
    (* proc *)  INIGet, INIPut;

FROM STextIO IMPORT
    (* proc *)  WriteString, WriteLn;

FROM SLongIO IMPORT
    (* proc *)  WriteReal, WriteFloat, WriteFixed;

FROM Names IMPORT
    (* type *)  ServiceType;

FROM TaskControl IMPORT
    (* type *)  Lock,
    (* proc *)  CreateTask, CreateLock, Obtain, Release;

FROM Timer IMPORT
    (* proc *)  Sleep;

FROM LowLevel IMPORT
    (* proc *)  EVAL;

(************************************************************************)
(*                                                                      *)
(* In module Names ServiceType is defined as (POP, SMTP, MSA, IMAP).    *)
(* This module doesn't record any IMAP information, so to keep the      *)
(* type structure simple we'll use the "IMAP" service type to refer to  *)
(* outgoing mail.                                                       *)
(*                                                                      *)
(************************************************************************)

CONST
    Minutes = 60;           (* seconds *)
    Hours = 60 * Minutes;   (* seconds *)
    Days = 24 * Hours;      (* seconds *)

    (* Arrival rate calculations rely on a time window, and for now I   *)
    (* think that it might as well be a global constant.                *)

    W = 3*Days;

CONST
    MemName = "\SHAREMEM\WEASEL\STATS";
    MutexName = "\SEM32\WEASEL\STATS";

TYPE
    (* The fields in a state record are:                                *)
    (*  lastarrival the time of the last arrival event                  *)
    (*  meandelta   average, at that time, of the times between arrivals *)
    (*  lambdaest   latest estimate of arrival rate.                    *)

    StateType =    RECORD
                        lastarrival: CARDINAL;
                        lambdaest: LONGREAL;
                        muinvest: LONGREAL;
                    END (*RECORD*);

    StateArray = ARRAY ServiceType OF StateType;

VAR
    (* The state information is a property of the server as a whole.    *)
    (* We can therefore afford to keep it as a global variable.         *)

    state: StateArray;

    (* Address of the shared memory where we store the state.  *)

    BaseAddr: POINTER TO StateArray;

    (* Critical section protection for shared memory access.  *)

    hmtx: OS2.HMTX;

    (* Critical section protection for INI file access. *)

    access: Lock;

    (* Flag that enables these calculations, and flag to say that       *)
    (* shared memory has been successfully allocated.                   *)

    statsenable, sharedMemOK: BOOLEAN;

    (* Flag to show shutdown in progress. *)

    Shutdown: BOOLEAN;

(************************************************************************)
(*                  NOTES ON THE ESTIMATION ALGORITHMS                  *)
(************************************************************************)
(*                                                                      *)
(*                   ARRIVAL RATE: WINDOWED APPROACH                    *)
(*                                                                      *)
(* I am using a windowed approach, where we count the arrivals in the   *)
(* last W time units.  That means the estimate of the arrival rate      *)
(* lambda at time t is lambda(t) = count(t)/W.  An exponentially        *)
(* smoothed estimate might be better, and maybe I'll look at that later,*)
(* but for now the windowed approach gives the simplest calculations.   *)
(* Let t_a be the time of the last arrival, and t_b be the time of      *)
(* the next arrival.  Initially we should set t_a to a large negative   *)
(* number, and count(t_a) = 0.  From then on,                           *)
(*              count(t_a) = lambda(t_a) * W                            *)
(* where lambda(t) is the best available estimate of the true lambda at *)
(* time t.  Of course this estimate is an approximation, because we     *)
(* are dealing with random variables, but the approximation should be   *)
(* a good one if W is large enough.                                     *)
(*                                                                      *)
(* Assume (for now) that t_b - t_a < W.  Then the window at time t_b    *)
(* overlaps the old window.  The amount of overlap, in time units, is   *)
(*              W - (t_b - t_a)                                         *)
(* and the expected count in that overlap period is lambda(t_a) times   *)
(* that overlap time.  At time t_b the count increases by one.  Thus    *)
(* our best estimate of the new count is                                *)
(*              lambda(t_a) * (W - (t_b - t_a)) + 1                     *)
(* (Of course we could get a more precise estimate by retaining a       *)
(* record of all past arrival times, but I am trying to avoid having to *)
(* keep too much state information.)  We then get a new estimate        *)
(* lambda(t_b) by dividing the new count by W.                          *)
(*                                                                      *)
(* In the case t_b - t_a > W the new window does not overlap the old,   *)
(* so count_t_b) = 1 and lambda(t_b) = 1/W.  This is equivalent to      *)
(* restarting the whole estimator from scratch, a situation we can      *)
(* avoid by making W large enough that a zero count in a whole window   *)
(* is improbable.  We should set W >> 1/lambda.                         *)
(*                                                                      *)
(* In the (improbable) case t_b - t_a = W, the guess lambda(t_b) = 2/W  *)
(* is as good as any.                                                   *)
(*                                                                      *)
(*                        FILTERED ESTIMATE                             *)
(*                                                                      *)
(* An alternative approach would be to set                              *)
(*      smoothed estimate := alpha*(old estimate)                       *)
(*                              + (1-alpha)*(new estimate)              *)
(* This is not quite a linear filter, because a linear filter has fixed *)
(* time steps, and we have random time steps.  It could be made into    *)
(* a linear filter by making alpha a function of the time step.  I'll   *)
(* set aside that complication for now.                                 *)
(* The obvious choice of the new estimate is                            *)
(*          new estimate = 1 / (t_b - t_a)                              *)
(* In separate tests this wasn't giving good results, so at least for   *)
(* now I've abandoned that approach, although I probably will use it    *)
(* for the service time estimator.                                      *)
(*                                                                      *)
(************************************************************************)

(************************************************************************)
(*                      ESTIMATOR FOR THE ARRIVAL RATE                  *)
(************************************************************************)

PROCEDURE UpdateEstimate (service: ServiceType;  now: CARDINAL);

    (* Calculates a new state[service].lambdaest. *)

    CONST alpha = 0.9;
        oneoverW = 1.0/LFLOAT(W);

    VAR lambda, newlambda: LONGREAL;
        step: CARDINAL;

    BEGIN
        lambda := state[service].lambdaest;
        step := now - state[service].lastarrival;
        IF step < W THEN
            newlambda := lambda * LFLOAT(W - step + 1) * oneoverW;
            state[service].lambdaest := newlambda;
        ELSIF step > W THEN
            state[service].lambdaest := oneoverW;
        ELSE
            state[service].lambdaest := 2.0 * oneoverW;
        END (*IF*);
        (*
        WriteString ("  ");
        WriteFixed (state[service].lambdaest, 4, 8);
        *)
    END UpdateEstimate;

(************************************************************************)

PROCEDURE RecordArrival (service: ServiceType);

    (* Records that a new arrival event happened at the current time.   *)

    VAR now: CARDINAL;

    BEGIN
        IF statsenable THEN
            Obtain (access);
            now := time();
            UpdateEstimate (service, now);
            state[service].lastarrival := now;
            Release (access);
        END (*IF*);
    END RecordArrival;

(************************************************************************)
(*                   ESTIMATOR FOR THE MEAN SERVICE TIME                *)
(************************************************************************)

PROCEDURE RecordServiceTime (s: ServiceType;  ms: CARDINAL);

    (* In this case the argument is in milliseconds, but we record mean *)
    (* service time in floating point seconds.                          *)

    CONST alpha = 0.95;

    VAR oldmean: LONGREAL;

    BEGIN
        IF statsenable THEN
            Obtain (access);
            oldmean := state[s].muinvest;
            state[s].muinvest := alpha*oldmean + (1.0-alpha)*0.001*LFLOAT (ms);
            Release (access);
        END (*IF*);
    END RecordServiceTime;

(************************************************************************)
(*                    WRITING DATA TO SHARED MEMORY                     *)
(************************************************************************)

PROCEDURE AllocateMemory(): BOOLEAN;

    (* Sets up the shared memory and its critical section protection.  *)

    VAR ulrc: CARDINAL;  success: BOOLEAN;
        name: ARRAY [0..63] OF CHAR;

    BEGIN
        (* Create shared memory. *)

        name := MemName;
        ulrc := OS2.DosAllocSharedMem (BaseAddr, name,
                    SIZE(StateArray),
                    OS2.PAG_COMMIT + OS2.PAG_READ + OS2.PAG_WRITE);
        success := ulrc = 0;

        (* Critical protection mutex.  *)

        IF success THEN
            name := MutexName;
            ulrc := OS2.DosCreateMutexSem (name, hmtx, OS2.DC_SEM_SHARED, FALSE);
            success := ulrc = 0;
        END (*IF*);

        RETURN success;

    END AllocateMemory;

(************************************************************************)

PROCEDURE DeallocateMemory;

    (* Frees the shared memory and its critical section protection.  We *)
    (* ignore error codes, because this procedure is called only when   *)
    (* when Weasel is shutting down.                                    *)

    VAR ulrc: CARDINAL;

    BEGIN
        ulrc := OS2.DosFreeMem (BaseAddr);
        ulrc := OS2.DosCloseMutexSem (hmtx);
    END DeallocateMemory;

(************************************************************************)

PROCEDURE StoreData;

    (* Runs as a separate task, periodically saving the data    *)
    (* to shared memory.                                        *)

    CONST sleeptime = 5000; (* milliseconds *)  (* make this longer for the production version? *)

    BEGIN
        WHILE NOT Shutdown DO
            Sleep (sleeptime);
            OS2.DosWaitEventSem (hmtx, OS2.SEM_INDEFINITE_WAIT);
            BaseAddr^ := state;
            OS2.DosPostEventSem (hmtx);
        END (*WHILE*);
    END StoreData;

(************************************************************************)
(*               LOADING/STORING OUR DATA IN THE INI FILE               *)
(*                                                                      *)
(*  We load the data from the INI file when Weasel starts up, and store *)
(*  it to the INI file on shutdown.  For the intervening time, we save  *)
(*  the results in shared memory, not in the INI file, because saving   *)
(*  to the INI file can cause clashes with Setup.  Setup will get the   *)
(*  data from the INI file only in two cases: when Weasel is not        *)
(*  running, or when using remote configuration.                        *)
(*                                                                      *)
(************************************************************************)

PROCEDURE LoadINIData (hini: INIData.HINI);

    (* Loads the estimates saved from the last run. *)

    VAR SYSapp: ARRAY [0..5] OF CHAR;
        s: ServiceType;

    BEGIN
        Obtain (access);
        SYSapp := "$SYS";
        IF NOT INIGet (hini, SYSapp, "statsenable", statsenable) THEN
            statsenable := FALSE;
        END (*IF*);
        IF NOT INIGet (hini, SYSapp, "stats", state) THEN
            FOR s := MIN(ServiceType) TO MAX(ServiceType) DO
                state[s].lambdaest := 0.0;
                state[s].muinvest := 0.0;
            END (*FOR*);
        END (*IF*);

        (* If the system is shut down and later restarted, the          *)
        (* lastarrival times can be misleading.  We need to reset the   *)
        (* arrival time to ensure that the shutdown period does not     *)
        (* influence the arrival rate calculation.                      *)

        FOR s := MIN(ServiceType) TO MAX(ServiceType) DO
            state[s].lastarrival := 0;
        END (*FOR*);
        Release (access);

        sharedMemOK := AllocateMemory();
        IF statsenable AND sharedMemOK THEN
            EVAL (CreateTask (StoreData, 4, "savestats"));
        END (*IF*);

    END LoadINIData;

(************************************************************************)

PROCEDURE StoreINIData;

    (* Writes the estimates back to disk. *)

    VAR hini: INIData.HINI;
        SYSapp: ARRAY [0..5] OF CHAR;

    BEGIN
        Obtain (access);
        SYSapp := "$SYS";
        hini := OpenINI();
        IF INIData.INIValid (hini) THEN
            INIPut (hini, SYSapp, "stats", state);
            CloseINI;
        END (*IF*);
        Release (access);
    END StoreINIData;

(************************************************************************)

VAR s: ServiceType;

BEGIN
    Shutdown := FALSE;
    statsenable := FALSE;
    FOR s := MIN(ServiceType) TO MAX(ServiceType) DO
        state[s].lastarrival := 0;
        state[s].lambdaest := 0.0;
        state[s].muinvest := 0.0;
    END (*FOR*);
    CreateLock (access);
    BaseAddr := NIL;  hmtx := 0;
FINALLY
    Shutdown := TRUE;
    IF sharedMemOK THEN
        DeallocateMemory;
    END (*IF*);
    StoreINIData;
END Stats.

