main.c

PB 熟悉的哥们希望大家可以互相学习一下

    rc = mco_trans_start(db, MCO_READ_WRITE, MCO_TRANS_FOREGROUND, &t);
    if (MCO_S_OK != rc)
    {
        printf("\nError starting trn\n");
        return rc;
    }

    rc = Rec_new(t, &rec);
    if (rc)
    {
        goto Err;
    }

    rc = Rec_key_put(&rec, key);
    if (rc)
    {
        goto Err;
    }

    vl = rand2(1, MAX_VECT_LEN);
    if ((rc = Rec_valuev_alloc(&rec, vl)))
    {
        goto Err;
    }

    for (i = 0; i < vl; i++)
    {

        if ((rc = Rec_valuev_put(&rec, i, &sv)))
        {
            goto Err;
        }
        if ((rc = Svalue_tm_put(&sv, i + idx)))
        {
            goto Err;
        }
        if ((rc = Svalue_u4_put(&sv, i + idx)))
        {
            goto Err;
        }
        str = get_random_string();
        len = (uint2)strlen(str);
        rc = Svalue_bytes_put(&sv, str, len);
    }

    sprintf(buf, "str %d", key);
    buflen = (uint2)strlen(buf);
    Rec_str_put(&rec, buf, buflen);

    //  recsum = record_checksum(key, buf, buflen);
    recsum = record_checksum_byhandle(&rec);

    chk->checksum_before_commit = chk->checksum_after_commit ^ recsum;

    rc = mco_trans_commit(t);
    if (rc == MCO_S_OK)
    {
        chk->checksum_after_commit = chk->checksum_before_commit;
    }
    if (rc)
    {
        i++;
    }
    idx += 0x123456;
    return rc;
    Err: mco_trans_rollback(t);
    idx += 0x123456;
    return rc;
}


static MCO_RET deleteTrn(mco_db_h db)
{
    MCO_RET rc;
    mco_trans_h t;
    Rec rec;
    CheckSums* chk = (CheckSums*)CHECKSUM_ADDR;
    int recsum;
    mco_cursor_t csr;

    // Delete first record

    rc = mco_trans_start(db, MCO_READ_WRITE, MCO_TRANS_FOREGROUND, &t);
    if (MCO_S_OK != rc)
    {
        printf("\nError starting trn\n");
        return rc;
    }

    Rec_tkey_index_cursor(t, &csr);
    rc = mco_cursor_first(t, &csr);
    if (rc != MCO_S_OK)
    {
        // no records
        goto Err;
    }
    rc = Rec_from_cursor(t, &csr, &rec);
    if (rc)
    {
        goto Err;
    }

    recsum = record_checksum_byhandle(&rec);

    rc = Rec_delete(&rec);
    if (rc)
    {
        goto Err;
    }
    chk->checksum_before_commit = chk->checksum_after_commit ^ recsum;
    rc = mco_trans_commit(t);
    if (rc == MCO_S_OK)
    {
        chk->checksum_after_commit = chk->checksum_before_commit;
    }
    return rc;

    Err: mco_trans_rollback(t);
    return rc;
}

/****************************************************
 *          "Check Resources" thread 
 ****************************************************/

THREAD_PROC_DEFINE(CheckResources, arg)
{
    mco_db_h db = 0;
    unsigned wdt_state = 0;

    THREAD_PROC_MODE();

    printf("check time %ld\n", CheckTime);
    Sleep(CheckTime);

    for (;;)
    {

        if (stopflag)
        {
            l10: stopflag--;
            printf("CheckResources thread complete\n");
            if (db != 0)
            {
                mco_db_disconnect(db);
            }
            return ;
        }

        Sleep(CheckTime);
        if (stopflag)
        {
            goto l10;
        }

        if (!startflag)
        {
            continue;
        }

        if (!mco_db_check_resources(dbName, &db, 40000, &wdt_state))
        {
            printf("\n*** recovery was done ***\n");
        }
    }

}


/****************************************************
 *          "Working" thread 
 ****************************************************/

THREAD_PROC_DEFINE(Working, par)
{
    int creating_db = 0; //1;
    MCO_RET rc;
    int cnt = 0;

    int ndeletes = 0, nfailed_deletes = 0, ninserts = 0, nfailedinserts = 0;

    THREAD_PROC_MODE();

    startflag = 0;
    Sleep(100);

    printf("\n****** Thread started ******\n");

    startflag++;

    creating_db = starting_create_new_db(&db);

    if (creating_db)
    {
        printf("\nCreating DB for the first time\n");
    }
    else
    {
        printf("\nRecovering DB after power off\n");
    }

    Sleep(500);

    rc = connect_to_db(&db, creating_db);

    if (rc)
    {
        printf("\nDb connecting error\n");
        for (;;)
            ;
    }

    printf("\nStarting cycle of inserts and deletes\n");

    for (cnt = 0; !stopflag; cnt++)
    // just creating & deleting records, updating checksum, until interrupted
    {
        int n;

        n = nRecords(db);

        count++;

        if (n > (20+cnt % 10))
        {
            ndeletes++;
            rc = deleteTrn(db);
            if (rc)
            {
                nfailed_deletes++;
            }
        }
        else
        {
            ninserts++;
            rc = insertTrn(db);

            if (rc)
            {
                nfailedinserts++;
            }
        }

        if (cnt % LOG_PERIOD == 0)
        {
            CheckSums* chk = (CheckSums*)CHECKSUM_ADDR;
            classStat(db);
            printf("\nCurrent Checksum = %d (%d)\n", chk->checksum_after_commit, chk->checksum_before_commit);

            printf("\n ndeletes=%d, nfailed_deletes=%d, ninserts=%d, nfailedinserts=%d\n", ndeletes, nfailed_deletes,
                   ninserts, nfailedinserts);
            Sleep(100);
        }

    }
    printf("\nWorking thread complete\n");
    if (db != 0)
    {
        mco_db_disconnect(db);
    }

    stopflag--;
}


void PowerOff()
{
    int i, j;

    startflag = 0;

    THREAD_CANCEL(thWorkThread);

    Sleep(100);

    printf("\n**************** RESTART *******************\n");
    printf("\n**************** RESTART *******************\n");
    printf("\n**************** RESTART *******************\n");

    for (i = 0; i < 100; i++)
    {
        if (stopflag)
        {
            goto l10;
        }
        j = count;
        Sleep(100);
        printf("\n*** Wait stopping working thread ***\n");
        if (count == j)
        {
            break;
        }
        count = 0;
    }
    printf("\n*** Restart working thread ***\n");

    THREAD_PROC_START(Working, 0, &thWorkThread);

    l10: ;
}

/* parsing command line */

int parse_cmd(int argc, char** argv)
{
    int i;
    char* p;
    int flag = 1;

    if (argc < 2)
    {
        return 1;
    }

    for (i = 1; i < argc; i++)
    {

        p = argv[i];

        if (strstr(p, "--test"))
        {
            test_Mode = 1;
            continue;
        }
        if ((p[0] == '/') || (p[0] == '-'))
        {
            if (p[1] == 'h')
            {
                help();
                return 0;
            }
            if (p[1] == 't')
            {
                flag++;
                if ((CheckTime = atoi((const char*) &p[2])) == 0)
                {
                    CheckTime = 1000;
                }
            }
        }
    }
    return flag;
}

#ifndef WIN32_WCE
    int main(int argc, char** argv)
#else 
    int __cdecl main(int argc, char** argv)
#endif 
{
    int i;
    FILE* f = stdin;
    char s[256];

    pdb = &db;

    PROCESS_MASKS(); // for LINUX, QNX, SOLARIS

    #if defined(_LINUX) || defined(_QNX) || defined(_SOLARIS) 
        sigdelset(&set, SIGALRM);
        sigprocmask(SIG_SETMASK, &set, 0);
    #endif 

    s[0] = 0;

    {
        mco_runtime_info_t info;

        mco_get_runtime_info(&info);

        if (!info.mco_recovery_supported)
        {
            printf("\n\n Recovery support is not configured! \n");
            PROG_EXIT(0);
        }
    }

    if (parse_cmd(argc, argv) == 0)
    {
        #ifdef _WIN32_WCE
            char str[128];
            printf("\npress ENTER to exit\n");
            gets(str);
        #endif 
        PROG_EXIT(0);
    }

    _SH_();

    mco_runtime_start();

    mco_error_set_handler(&errhandler);

    mco_runtime_setoption(MCO_RT_OPTION_ZMUTEX_NR1,  // loop  waiting the resource
    long_loop  // the loop value
    );

    make_strings();

    if (CheckTime)
    {
        THREAD_PROC_START(CheckResources, 0, &thCheckRes); /* run "Check Resources" thread */
    }

    THREAD_PROC_START(Working, 0, &thWorkThread);

    Sleep(1000);

    for (;;)
    {
        #ifdef _VXWORKS
            i = 0;
            while (i == 0)
            {
                ioctl(f->_file, FIONREAD, (int) &i);
                Sleep(100);
            }
        #endif 
        if (test_Mode == 1)
        {
            Sleep(5000);
        }

        if ((fgets(s, 256, stdin)) == 0)
        {
            continue;
        }
        if (s[0] == 0)
        {
            continue;
        }
        i = atoi(s);
        if (i == 343)
        {
            EXIT(1);
        }
        if (s[0] == 'e')
        {
            stopflag = 2;
            break;
        }

        PowerOff();
    }

    Printf("\n\nApplication interrupted by keystroke\n");
    Printf("Wait for the completion of working threads\n");

    for (i = 0; i < 30; i++)
    {

        if (!stopflag)
        {
            break;
        }

        //      Printf("%d running threads lost\n", stopflag);
        Sleep(100); /* wait for terminating of working threads */
    }


    //    mco_db_disconnect(db);
    mco_db_close(dbName);

    mco_runtime_stop();

    printf("program termination\n");

    PROG_EXIT(0);

}