oracle_txns8.c

oracle数据库tpcc(在线事务处理能力）测试的源码

#define SQLTXT4 "UPDATE /*+ buffer */ ordl SET ol_delivery_d =:cr_date \
	WHERE ol_w_id=:w_id and ol_d_id=:d_id and ol_o_id=:o_id \
	RETURNING ol_amount into :ol_amount "
#define SQLTXT6 "UPDATE cust SET c_balance = c_balance + :amt, \
	c_delivery_cnt = c_delivery_cnt + 1 WHERE c_w_id = :w_id AND \
	c_d_id = :d_id AND c_id = :c_id"
sb4 no_data(dvoid *ctxp, OCIBind *bp, ub4 iter, ub4 index,
	dvoid **bufpp, ub4 *alenp, ub1 *piecep,
	dvoid **indpp)
{
   *bufpp = (dvoid*)0;
   *alenp =0;
   *indpp = (dvoid*)0;
   *piecep =OCI_ONE_PIECE;
   return (OCI_CONTINUE);
}
sb4 TPC_oid_data(dvoid *ctxp, OCIBind *bp, ub4 iter, ub4 index,
	dvoid **bufpp, ub4 **alenp, ub1 *piecep,
	dvoid **indpp, ub2 **rcodepp)
{
   delctx *dctx;
   dctx=(delctx*)ctxp;
   *bufpp = &dctx->del_o_id[iter];
   *indpp= &dctx->del_o_id_ind[iter];
   dctx->del_o_id_len[iter]=sizeof(dctx->del_o_id[0]);
   *alenp= &dctx->del_o_id_len[iter];
   *rcodepp = &dctx->del_o_id_rcode[iter];
   *piecep =OCI_ONE_PIECE;
    return (OCI_CONTINUE);
}
sb4 cid_data(dvoid *ctxp, OCIBind *bp, ub4 iter, ub4 index,
	dvoid **bufpp, ub4 **alenp, ub1 *piecep,
	dvoid **indpp, ub2 **rcodepp)
{
    delctx *dctx;
    dctx=(delctx*)ctxp;
    *bufpp = &dctx->c_id[iter];
    *indpp= &dctx->c_id_ind[iter];
    dctx->c_id_len[iter]=sizeof(dctx->c_id[0]);
    *alenp= &dctx->c_id_len[iter];
    *rcodepp = &dctx->c_id_rcode[iter];
    *piecep =OCI_ONE_PIECE;
    return (OCI_CONTINUE);
}
sb4 amt_data(dvoid *ctxp, OCIBind *bp, ub4 iter, ub4 index,
	dvoid **bufpp, ub4 **alenp, ub1 *piecep,
	dvoid **indpp, ub2 **rcodepp)
{
   amtctx *actx;
   actx =(amtctx*)ctxp;
   actx->ol_cnt[iter]=actx->ol_cnt[iter]+1;
   *bufpp = &actx->ol_amt[iter][index];
   *indpp= &actx->ol_amt_ind[iter][index];
   actx->ol_amt_len[iter][index]=sizeof(actx->ol_amt[0][0]);
   *alenp= &actx->ol_amt_len[iter][index];
   *rcodepp = &actx->ol_amt_rcode[iter][index];
   *piecep =OCI_ONE_PIECE;
   return (OCI_CONTINUE);
}

void shiftdata(from,dctx)
int from;
delctx *dctx;
{
   int i;
   for (i=from;i<NDISTS-1; i++)
   {
      dctx->del_o_id_ind[i] = dctx->del_o_id_ind[i+1];
      dctx->del_o_id[i] = dctx->del_o_id[i+1];
      dctx->w_id[i] = dctx->w_id[i+1];
      dctx->d_id[i] = dctx->d_id[i+1];
      dctx->carrier_id[i] = dctx->carrier_id[i+1];
   }
} /* end shiftdata */

tkvcdinit (DeliveryData *pDel,OraContext *p)
{
   delctx *dctx = &(p->dctx);
   amtctx *actx = &(p->actx);
   text stmbuf[8192];
   memset(dctx,(char)0,sizeof(delctx));
   memset(actx,(char)0,sizeof(amtctx));
   dctx->norow = 0;
   dctx->curd1 = NULL;
   dctx->curd2 = NULL;
   dctx->curd3 = NULL;
   dctx->curd4 = NULL;
   dctx->curd5 = NULL;
   dctx->curd6 = NULL;
   #if defined(ISO) || defined(ISO5) || defined(ISO6) || defined(ISO8)
       OCIHandleAlloc(tpcenv, (dvoid **)(&dctx->curd0), OCI_HTYPE_STMT,0, (dvoid**)0);
   sprintf ((char *) stmbuf, SQLTXT0);
   OCIStmtPrepare(dctx->curd0, p->errhp, stmbuf, strlen((char*)stmbuf),OCI_NTV_SYNTAX, OCI_DEFAULT);
   OCIDFNRA(dctx->curd0, dctx->inum_dp,p->errhp,1,dctx->inum,SIZ(dctx->inum),SQLT_STR,
	dctx->inum_ind,dctx->inum_len,dctx->inum_rcode);
   #endif
/* open first cursor */
   OCIHandleAlloc(p->tpcenv, (dvoid **)(&dctx->curd1),OCI_HTYPE_STMT,0, (dvoid**)0);
   sprintf ((char *) stmbuf, SQLTXT1);
   OCIStmtPrepare(dctx->curd1, p->errhp, stmbuf, strlen((char*)stmbuf),OCI_NTV_SYNTAX, OCI_DEFAULT);
/* bind variables */
   OCIBND(dctx->curd1, dctx->w_id_bp,p,":w_id",dctx->w_id,SIZ(int),SQLT_INT);
   OCIBNDRA(dctx->curd1, dctx->d_id_bp,p,":d_id",dctx->d_id,SIZ(int),
	SQLT_INT,NULL,NULL,NULL);
   OCIBNDRAD(dctx->curd1, dctx->del_o_id_bp,p,":o_id",SIZ(int),
	SQLT_INT,NULL,dctx,no_data,TPC_oid_data);
/* open third cursor */
   OCIHandleAlloc(p->tpcenv, (dvoid **)(&dctx->curd3),OCI_HTYPE_STMT, 0,(dvoid**)0);
   sprintf ((char *) stmbuf, SQLTXT3);
   OCIStmtPrepare(dctx->curd3, p->errhp, stmbuf, strlen((char*)stmbuf),
	OCI_NTV_SYNTAX, OCI_DEFAULT);
/* bind variables */
   OCIBNDRA(dctx->curd3, dctx->carrier_id_bp,p,":carrier_id",dctx->carrier_id,
	SIZ(int),SQLT_INT,dctx->carrier_id_ind,
	dctx->carrier_id_len,dctx->carrier_id_rcode);
   OCIBNDRA(dctx->curd3, dctx->w_id_bp3,p,":w_id",dctx->w_id,
	SIZ(int),SQLT_INT,NULL,NULL,NULL);
   OCIBNDRA(dctx->curd3, dctx->d_id_bp3,p,":d_id",dctx->d_id,
	SIZ(int),SQLT_INT,NULL,NULL,NULL);
   OCIBNDRA(dctx->curd3, dctx->del_o_id_bp3,p,":o_id",dctx->del_o_id,
	SIZ(int),SQLT_INT,NULL,NULL,NULL);
   OCIBNDRAD(dctx->curd3, dctx->c_id_bp3,p,":o_c_id",SIZ(int),
	SQLT_INT,NULL,dctx,no_data,cid_data);
/* open fourth cursor */
   OCIHandleAlloc(p->tpcenv, (dvoid **)(&dctx->curd4),OCI_HTYPE_STMT, 0,(dvoid**)0);
   sprintf ((char *) stmbuf, SQLTXT4);
   OCIStmtPrepare(dctx->curd4, p->errhp, stmbuf, strlen((char*)stmbuf),
	OCI_NTV_SYNTAX, OCI_DEFAULT);
/* bind variables */
   OCIBND(dctx->curd4, dctx->w_id_bp4,p,":w_id",dctx->w_id,SIZ(dctx->w_id[0]),SQLT_INT);
   OCIBND(dctx->curd4, dctx->d_id_bp4,p,":d_id",dctx->d_id,SIZ(dctx->d_id[0]),SQLT_INT);
   OCIBND(dctx->curd4, dctx->o_id_bp,p,":o_id",dctx->del_o_id,SIZ(int),SQLT_INT);
   OCIBND(dctx->curd4, dctx->cr_date_bp,p,":cr_date",dctx->del_date,SIZ(OCIDate), SQLT_ODT);
   OCIBNDRAD(dctx->curd4, dctx->olamt_bp,p,":ol_amount",SIZ(int),
	SQLT_INT,NULL,actx,no_data,amt_data);
/* open sixth cursor */
   OCIHandleAlloc(p->tpcenv, (dvoid **)(&dctx->curd6),OCI_HTYPE_STMT, 0,(dvoid**)0);
   sprintf ((char *) stmbuf, SQLTXT6);
   OCIStmtPrepare(dctx->curd6, p->errhp, stmbuf, strlen((char*)stmbuf),
	OCI_NTV_SYNTAX, OCI_DEFAULT);
/* bind variables */
   OCIBND(dctx->curd6, dctx->amt_bp,p,":amt",dctx->amt,SIZ(int),SQLT_INT);
   OCIBND(dctx->curd6, dctx->w_id_bp6,p,":w_id",dctx->w_id,SIZ(int),SQLT_INT);
   OCIBND(dctx->curd6, dctx->d_id_bp6,p,":d_id",dctx->d_id,SIZ(int),SQLT_INT);
   OCIBND(dctx->curd6, dctx->c_id_bp,p,":c_id",dctx->c_id,SIZ(int),SQLT_INT);
   return (ERR_DB_SUCCESS);
}

tkvcd (DeliveryData *pDel, OraContext *p)
{
   delctx *dctx = &(p->dctx);
   amtctx *actx = &(p->actx);
   deltemp *dtemp = &(p->tempvars.del);
   int i, j;
   int rpc,rcount,count;
   int invalid;
   unsigned char localcr_date[7];
   OCIError *datecvterrhp = p->datecvterrhp;
   int execstatus;
   int errcode;
   int proc_no = 0;
   invalid = 0;
   vgetdate(localcr_date);
   cvtdmyhms(localcr_date,dtemp->cvtcr_date);
   OCIDateFromText(datecvterrhp,dtemp->cvtcr_date,strlen(dtemp->cvtcr_date),
   	"DD-MM-YYYY HH24:MI:SS",21,(text *) 0, 0,&dtemp->cr_date);
   #if defined(ISO) || defined(ISO5) || defined(ISO6) || defined(ISO8)
	int hasno;
	int reread;
	char sdate[30];
	OCIStmtExecute(p->tpcsvc,dctx->curd0,p->errhp,1,0,0,0,OCI_DEFAULT);
	sysdate(sdate);
	printf ("Delivery started at %s on node %s\n",sdate,dctx->inum);
   #endif
   #if defined(ISO) || defined(ISO5) || defined(ISO6) || defined(ISO8)
	reread = 1;
   #endif
   #if defined(ISO) || defined(ISO5) || defined(ISO6) || defined(ISO8)
	iso:
   #endif
/* initialization for array operations */
   for (i = 0; i < NDISTS; i++) 
   {
      dctx->del_o_id_ind[i] = TRUE;
      dctx->cons_ind[i] = TRUE;
      dctx->w_id_ind[i] = TRUE;
      dctx->d_id_ind[i] = TRUE;
      dctx->c_id_ind[i] = TRUE;
      dctx->del_date_ind[i] = TRUE;
      dctx->carrier_id_ind[i] = TRUE;
      dctx->amt_ind[i] = TRUE;
      dctx->no_rowid_ind[i] = TRUE;
      dctx->o_rowid_ind[i] = TRUE;
      dctx->del_o_id_len[i] = SIZ(dctx->del_o_id[0]);
      dctx->cons_len[i] = SIZ(dctx->cons[0]);
      dctx->w_id_len[i] = SIZ(dctx->w_id[0]);
      dctx->d_id_len[i] = SIZ(dctx->d_id[0]);
      dctx->c_id_len[i] = SIZ(dctx->c_id[0]);
      dctx->del_date_len[i] = DEL_DATE_LEN;
      dctx->carrier_id_len[i] = SIZ(dctx->carrier_id[0]);
      dctx->amt_len[i] = SIZ(dctx->amt[0]);
      dctx->no_rowid_len[i] = ROWIDLEN;
      dctx->o_rowid_len[i] = ROWIDLEN;
      dctx->o_rowid_ptr_len[i] = SIZ(dctx->o_rowid_ptr[0]);
      dctx->no_rowid_ptr_len[i] = SIZ(dctx->no_rowid_ptr[0]);
      dctx->w_id[i] = pDel->w_id;
      dctx->d_id[i] = i+1;
      dctx->carrier_id[i] = pDel->o_carrier_id;
      memcpy(&dctx->del_date[i],&dtemp->cr_date,sizeof(OCIDate));
      actx->ol_cnt[i]=0;
   }
/* array select from new_order and orders tables */
   execstatus=OCIStmtExecute(p->tpcsvc,dctx->curd1,p->errhp,NDISTS,0,0,0,OCI_DEFAULT);
   if((execstatus != OCI_SUCCESS) && (execstatus != OCI_NO_DATA))
   {
      OCITransRollback(p->tpcsvc,p->errhp,OCI_DEFAULT);
      errcode = OCIERROR(p,execstatus);
      if(errcode == NOT_SERIALIZABLE)
      {
         return(RECOVERR);
      }
      else if (errcode == RECOVERR)
      {
            return(RECOVERR);
      }
      else if (errcode == SNAPSHOT_TOO_OLD)
      {
         return(RECOVERR);
      }
      else
      {
          return (ERR_DB_ERROR);
      }
   }
/* mark districts with no new order */
   OCIAttrGet(dctx->curd1,OCI_HTYPE_STMT,&rcount,0,OCI_ATTR_ROW_COUNT,p->errhp);
   rpc = rcount;
   if (rcount != NDISTS)
   {
      int j=0;
      for (i=0;i<NDISTS;i++)
      {
	 if (dctx->del_o_id_ind[j] == 0) /* there is data here */
	    j++;
	 else 
	    shiftdata(j,dctx);
      }
   }
   #if defined(ISO) || defined(ISO5) || defined(ISO6) || defined(ISO8)
      if (invalid)
      {
	 sysdate(sdate);
	 for (i=1;i<=NDISTS;i++)
	 {
		hasno=0;
		for (j=0;j<rpc;j++)
		{
			if (dctx->d_id[j] == i)
			{
				hasno=1;
				break;
			}
		}
		if (!hasno)
	 	printf ("Delivery [dist %d] found no new order at %s\n",i,sdate);
	 }
	 if (reread)
	 {
		sleep (60);
		sysdate(sdate);
		printf ("Delivery wake up at %s\n",sdate);
		reread=0;
		goto iso;
	 }
      } /* end if (invalid) */
   #endif
   execstatus=OCIStmtExecute(p->tpcsvc,dctx->curd3,p->errhp,rpc,0,0,0,OCI_DEFAULT);
   if(execstatus != OCI_SUCCESS)
   {
      OCITransRollback(p->tpcsvc,p->errhp,OCI_DEFAULT);
      errcode = OCIERROR(p,execstatus);
      if(errcode == NOT_SERIALIZABLE)
      {
	 return(RECOVERR);
      }
      else if (errcode == RECOVERR)
      {
	 return (RECOVERR);
      }
      else if (errcode == SNAPSHOT_TOO_OLD)
      {
	 return (RECOVERR);
      }
      else
      {
	 return (ERR_DB_ERROR);
      }
   }
   OCIAttrGet(dctx->curd3,OCI_HTYPE_STMT,&rcount,0,OCI_ATTR_ROW_COUNT,p->errhp);
   if (rcount != rpc)
   {
      TPCCErr( "Error in TPC-C server %d: %d rows selected, %d ordsupdated\n",
		proc_no, rpc, rcount);
      OCITransRollback(p->tpcsvc,p->errhp,OCI_DEFAULT);
      return (ERR_DB_ERROR);
    }
/* array update of order_line table */
    execstatus=OCIStmtExecute(p->tpcsvc,dctx->curd4,p->errhp,rpc,0,0,0,OCI_DEFAULT);
   if(execstatus != OCI_SUCCESS)
   {
	OCITransRollback(p->tpcsvc,p->errhp,OCI_DEFAULT);
	errcode = OCIERROR(p,execstatus);
	if(errcode == NOT_SERIALIZABLE)
	{
		return(RECOVERR);
	}
	else if (errcode == RECOVERR)
	{
		return(RECOVERR);
	}
	else if (errcode == SNAPSHOT_TOO_OLD)
	{
		return(RECOVERR);
	}
	else
	{
		return (ERR_DB_ERROR);
	}
   }
   OCIAttrGet(dctx->curd4,OCI_HTYPE_STMT,&rcount,NULL,OCI_ATTR_ROW_COUNT,p->errhp);
/* add up amounts */
   count=0;
   for (i=0;i<rpc;i++)
   {
	dctx->amt[i]=0;
	for (j=0;j<actx->ol_cnt[i];j++)
	if (actx->ol_amt_rcode[i][j] == 0)
	{
	   dctx->amt[i] = dctx->amt[i] + actx->ol_amt[i][j];
	   count = count+1;
	}
   }
   if (rcount > rpc*NITEMS)
   {
	TPCCErr( "Error in TPC-C server %d: %d ordnrs updated, %dordl updated\n",
		proc_no, rpc, rcount);
   }