test_event_merge.cpp

mysql-5.0.22.tar.gz源码包

resetmem()
{
  int i, j;
  for (j = 0; j < 2; j++) {
    for (i = 0; i < g_maxcol; i++) {
      g_ev_ra[j][i] = 0;
      g_ev_bh[j][i] = 0;
    }
  }
  if (g_rec_ev != 0) {
    freeop(g_rec_ev);
    g_rec_ev = 0;
  }
  Uint32 pk1;
  for (pk1 = 0; pk1 < g_opts.maxpk; pk1++)
    g_ev_pos[pk1] = 0;
  // leave g_seq
  for (pk1 = 0; pk1 < g_opts.maxpk; pk1++) {
    if (g_pk_op[pk1] != 0) {
      Op* tot_op = g_pk_op[pk1];
      while (tot_op->next_gci != 0) {
        Op* gci_op = tot_op->next_gci;
        while (gci_op->next_com != 0) {
          Op* com_op = gci_op->next_com;
          while (com_op->next_op != 0) {
            Op* op = com_op->next_op;
            com_op->next_op = op->next_op;
            freeop(op);
          }
          gci_op->next_com = com_op->next_com;
          freeop(com_op);
        }
        tot_op->next_gci = gci_op->next_gci;
        freeop(gci_op);
      }
      freeop(tot_op);
      g_pk_op[pk1] = 0;
    }
    if (g_pk_ev[pk1] != 0) {
      Op* tot_op = g_pk_ev[pk1];
      while (tot_op->next_ev != 0) {
        Op* ev = tot_op->next_ev;
        tot_op->next_ev = ev->next_ev;
        freeop(ev);
      }
      freeop(tot_op);
      g_pk_ev[pk1] = 0;
    }
  }
  assert(g_usedops == 0);
}

struct Comp {
  Op::Type t1, t2, t3;
};

static Comp
g_comp[] = {
  { Op::INS, Op::DEL, Op::NUL },
  { Op::INS, Op::UPD, Op::INS },
  { Op::DEL, Op::INS, Op::UPD },
  { Op::UPD, Op::DEL, Op::DEL },
  { Op::UPD, Op::UPD, Op::UPD }
};

static const uint g_ncomp = sizeof(g_comp)/sizeof(g_comp[0]);

static int
checkop(const Op* op, Uint32& pk1)
{
  Op::Type t = op->type;
  if (t == Op::NUL)
    return 0;
  chkrc(t == Op::INS || t == Op::DEL || t == Op::UPD);
  const Data& d0 = op->data[0];
  const Data& d1 = op->data[1];
  {
    const Col& c = getcol("pk1");
    chkrc(d0.ind[c.no] == 0);
    pk1 = d0.pk1;
    chkrc(pk1 < g_opts.maxpk);
  }
  uint i;
  for (i = 0; i < ncol(); i++) {
    const Col& c = getcol(i);
    const int ind0 = d0.ind[i];
    const int ind1 = d1.ind[i];
    // the rules are the rules..
    if (c.pk) {
      chkrc(ind0 == 0); // always PK in post data
      if (t == Op::INS)
        chkrc(ind1 == -1);
      if (t == Op::DEL)
        chkrc(ind1 == -1); // no PK in pre data
      if (t == Op::UPD)
        chkrc(ind1 == 0);
    }
    if (! c.pk) {
      if (t == Op::INS)
        chkrc(ind0 >= 0 && ind1 == -1);
      if (t == Op::DEL)
        chkrc(ind0 == -1 && ind1 >= 0); // always non-PK in pre data
      if (t == Op::UPD)
        chkrc(ind0 == -1 || ind1 >= 0); // update must have pre data
    }
    if (! c.nullable) {
      chkrc(ind0 <= 0 && ind1 <= 0);
    }
  }
  return 0;
}

static Comp*
comptype(Op::Type t1, Op::Type t2) // only non-NUL
{
  uint i;
  for (i = 0; i < g_ncomp; i++)
    if (g_comp[i].t1 == t1 && g_comp[i].t2 == t2)
      return &g_comp[i];
  return 0;
}

static void
copycol(const Col& c, const Data& d1, Data& d3)
{
  uint i = c.no;
  if ((d3.ind[i] = d1.ind[i]) == 0) {
    if (! c.isblob()) {
      memmove(d3.ptr[i].v, d1.ptr[i].v, c.size);
    } else {
      Data::Txt& t1 = *d1.ptr[i].txt;
      Data::Txt& t3 = *d3.ptr[i].txt;
      delete [] t3.val;
      t3.val = new char [t1.len];
      t3.len = t1.len;
      memcpy(t3.val, t1.val, t1.len);
    }
  }
}

static void
copydata(const Data& d1, Data& d3, bool pk, bool nonpk)
{
  uint i;
  for (i = 0; i < ncol(); i++) {
    const Col& c = g_col[i];
    if (c.pk && pk || ! c.pk && nonpk)
      copycol(c, d1, d3);
  }
}

static void
compdata(const Data& d1, const Data& d2, Data& d3, bool pk, bool nonpk)
{
  uint i;
  for (i = 0; i < ncol(); i++) {
    const Col& c = g_col[i];
    if (c.pk && pk || ! c.pk && nonpk) {
      const Data* d = 0;
      if (d1.ind[i] == -1 && d2.ind[i] == -1)
        d3.ind[i] = -1;
      else if (d1.ind[i] == -1 && d2.ind[i] != -1)
        d = &d2;
      else if (d1.ind[i] != -1 && d2.ind[i] == -1)
        d = &d1;
      else
        d = &d2;
      if (d != 0)
        copycol(c, *d, d3);
    }
  }
}

static void
copyop(const Op* op1, Op* op3)
{
  op3->type = op1->type;
  copydata(op1->data[0], op3->data[0], true, true);
  copydata(op1->data[1], op3->data[1], true, true);
  op3->gci = op1->gci;
  Uint32 pk1_tmp;
  reqrc(checkop(op3, pk1_tmp) == 0);
}

static int
compop(const Op* op1, const Op* op2, Op* op3) // op1 o op2 = op3
{
  Comp* comp;
  if (op2->type == Op::NUL) {
    copyop(op1, op3);
    return 0;
  }
  if (op1->type == Op::NUL) {
    copyop(op2, op3);
    return 0;
  }
  Op::Kind kind =
    op1->kind == Op::OP && op2->kind == Op::OP ? Op::OP : Op::EV;
  Op* res_op = getop(kind);
  chkrc((comp = comptype(op1->type, op2->type)) != 0);
  res_op->type = comp->t3;
  if (res_op->type == Op::INS) {
    // INS o UPD
    compdata(op1->data[0], op2->data[0], res_op->data[0], true, true);
    // pre = undef
  }
  if (res_op->type == Op::DEL) {
    // UPD o DEL
    copydata(op2->data[0], res_op->data[0], true, false); // PK
    copydata(op1->data[1], res_op->data[1], false, true); // non-PK
  } 
  if (res_op->type == Op::UPD && op1->type == Op::DEL) {
    // DEL o INS
    copydata(op2->data[0], res_op->data[0], true, true);
    copydata(op1->data[0], res_op->data[1], true, false); // PK
    copydata(op1->data[1], res_op->data[1], false, true); // non-PK
  }
  if (res_op->type == Op::UPD && op1->type == Op::UPD) {
    // UPD o UPD
    compdata(op1->data[0], op2->data[0], res_op->data[0], true, true);
    compdata(op2->data[1], op1->data[1], res_op->data[1], true, true);
  }
  assert(op1->gci == op2->gci);
  res_op->gci = op2->gci;
  Uint32 pk1_tmp;
  reqrc(checkop(res_op, pk1_tmp) == 0);
  copyop(res_op, op3);
  freeop(res_op);
  return 0;
}

static int
createevent()
{
  ll1("createevent");
  g_evt = 0;
  g_dic = g_ndb->getDictionary();
  NdbDictionary::Event evt(g_evtname);
  evt.setTable(*g_tab);
  evt.addTableEvent(NdbDictionary::Event::TE_ALL);
  uint i;
  for (i = 0; i < ncol(); i++) {
    const Col& c = g_col[i];
    evt.addEventColumn(c.name);
  }
#ifdef version51rbr
  evt.separateEvents(g_opts.separate_events);
#endif
  if (g_dic->getEvent(evt.getName()) != 0)
    chkdb(g_dic->dropEvent(evt.getName()) == 0);
  chkdb(g_dic->createEvent(evt) == 0);
  chkdb((g_evt = g_dic->getEvent(evt.getName())) != 0);
  g_dic = 0;
  return 0;
}

static int
dropevent()
{
  ll1("dropevent");
  g_dic = g_ndb->getDictionary();
  chkdb(g_dic->dropEvent(g_evt->getName()) == 0);
  g_evt = 0;
  g_dic = 0;
  return 0;
}

static int
createeventop()
{
  ll1("createeventop");
#ifdef version50
  uint bsz = 10 * g_opts.maxops;
  chkdb((g_evt_op = g_ndb->createEventOperation(g_evt->getName(), bsz)) != 0);
#else
  chkdb((g_evt_op = g_ndb->createEventOperation(g_evt->getName())) != 0);
#ifdef version51rbr
  g_evt_op->separateEvents(g_opts.separate_events); // not yet inherited
#endif
#endif
  uint i;
  for (i = 0; i < ncol(); i++) {
    const Col& c = g_col[i];
    Data (&d)[2] = g_rec_ev->data;
    if (! c.isblob()) {
      chkdb((g_ev_ra[0][i] = g_evt_op->getValue(c.name, (char*)d[0].ptr[i].v)) != 0);
      chkdb((g_ev_ra[1][i] = g_evt_op->getPreValue(c.name, (char*)d[1].ptr[i].v)) != 0);
    } else {
#ifdef version51rbr
      chkdb((g_ev_bh[0][i] = g_evt_op->getBlobHandle(c.name)) != 0);
      chkdb((g_ev_bh[1][i] = g_evt_op->getPreBlobHandle(c.name)) != 0);
#endif
    }
  }
  return 0;
}

static int
dropeventop()
{
  ll1("dropeventop");
  chkdb(g_ndb->dropEventOperation(g_evt_op) == 0);
  g_evt_op = 0;
  return 0;
}

static int
waitgci() // wait for event to be installed and for at least 1 GCI to pass
{
  const uint ngci = 3;
  ll1("waitgci " << ngci);
  Uint32 gci[2];
  uint i = 0;
  while (1) {
    chkdb((g_con = g_ndb->startTransaction()) != 0);
    { // forced to exec a dummy op
      Uint32 pk1;
      char pk2[g_charlen + 1];
      pk1 = g_maxpk;
      sprintf(pk2, "%-*u", g_charlen, pk1);
      chkdb((g_op = g_con->getNdbOperation(g_tabname)) != 0);
      chkdb(g_op->readTuple() == 0);
      chkdb(g_op->equal("pk1", (char*)&pk1) == 0);
      chkdb(g_op->equal("pk2", (char*)&pk2[0]) == 0);
      chkdb(g_con->execute(Commit) == 0);
      g_op = 0;
    }
    gci[i] = g_con->getGCI();
    g_ndb->closeTransaction(g_con);
    g_con = 0;
    if (i == 1 && gci[0] + ngci <= gci[1]) {
      ll1("waitgci: " << gci[0] << " " << gci[1]);
      break;
    }
    i = 1;
    sleep(1);
  }
  return 0;
}

// scan table and set current tot_op for each pk1
static int
scantab()
{
  NdbRecAttr* ra[g_maxcol];
  NdbBlob* bh[g_maxcol];
  Op* rec_op = getop(Op::OP);
  Data& d0 = rec_op->data[0];
  chkdb((g_con = g_ndb->startTransaction()) != 0);
  chkdb((g_scan_op = g_con->getNdbScanOperation(g_tabname)) != 0);
  chkdb(g_scan_op->readTuples() == 0);
  uint i;
  for (i = 0; i < ncol(); i++) {
    const Col& c = getcol(i);
    if (! c.isblob()) {
      chkdb((ra[i] = g_scan_op->getValue(c.name, (char*)d0.ptr[i].v)) != 0);
    } else {
      chkdb((bh[i] = g_scan_op->getBlobHandle(c.name)) != 0);
    }
  }
  chkdb(g_con->execute(NoCommit) == 0);
  int ret;
  while ((ret = g_scan_op->nextResult()) == 0) {
    Uint32 pk1 = d0.pk1;
    if (pk1 >= g_opts.maxpk)
      continue;
    rec_op->type = Op::INS;
    for (i = 0; i < ncol(); i++) {
      const Col& c = getcol(i);
      int ind;
      if (! c.isblob()) {
        ind = ra[i]->isNULL();
      } else {
#ifdef version51rbr
        int ret;
        ret = bh[i]->getDefined(ind);
        assert(ret == 0);
        if (ind == 0) {
          Data::Txt& t = *d0.ptr[i].txt;
          Uint64 len64;
          ret = bh[i]->getLength(len64);
          assert(ret == 0);
          t.len = (uint)len64;
          delete [] t.val;
          t.val = new char [t.len];
          memset(t.val, 'X', t.len);
          Uint32 len = t.len;
          ret = bh[i]->readData(t.val, len);
          assert(ret == 0 && len == t.len);
        }
#endif
      }
      assert(ind >= 0);
      d0.ind[i] = ind;
    }
    assert(g_pk_op[pk1] == 0);
    Op* tot_op = g_pk_op[pk1] = getop(Op::OP);
    copyop(rec_op, tot_op);
    tot_op->type = Op::INS;
  }
  chkdb(ret == 1);
  g_ndb->closeTransaction(g_con);
  g_scan_op = 0;
  g_con = 0;
  freeop(rec_op);
  return 0;
}

static void
makedata(const Col& c, Data& d, Uint32 pk1, Op::Type t)
{
  uint i = c.no;
  if (c.pk) {
    switch (c.type) {
    case NdbDictionary::Column::Unsigned:
      {
        Uint32* p = d.ptr[i].u32;
        *p = pk1;
      }
      break;
    case NdbDictionary::Column::Char:
      {
        char* p = d.ptr[i].ch;
        sprintf(p, "%-*u", g_charlen, pk1);
      }
      break;
    default:
      assert(false);
      break;
    }
    d.ind[i] = 0;
  } else if (t == Op::DEL) {
    ;
  } else if (i == getcol("seq").no) {
    d.seq = g_seq++;
    d.ind[i] = 0;
  } else if (t == Op::INS && c.nullable && urandom(10, 100)) {
    d.noop |= (1 << i);
    d.ind[i] = 1; // implicit NULL value is known
  } else if (t == Op::UPD && urandom(10, 100)) {
    d.noop |= (1 << i);
    d.ind[i] = -1; // fixed up in caller
  } else if (c.nullable && urandom(10, 100)) {
    d.ind[i] = 1;
  } else {
    switch (c.type) {
    case NdbDictionary::Column::Unsigned:
      {
        Uint32* p = d.ptr[i].u32;
        uint u = urandom();
        *p = u;
      }
      break;
    case NdbDictionary::Column::Char:
      {
        char* p = d.ptr[i].ch;
        uint u = urandom(g_charlen);
        uint j;
        for (j = 0; j < g_charlen; j++) {
          uint v = urandom(strlen(g_charval));
          p[j] = j < u ? g_charval[v] : 0x20;
        }
      }
      break;
    case NdbDictionary::Column::Text:
      {
        Data::Txt& t = *d.ptr[i].txt;
        uint u = urandom(g_maxblobsize);
        u = urandom(u); // 4x bias for smaller blobs
        u = urandom(u);
        delete [] t.val;
        t.val = new char [u];
        t.len = u;
        uint j = 0;
        while (j < u) {
          assert(u > 0);
          uint k = 1 + urandom(u - 1);
          if (k > u - j)
            k = u - j;
          uint v = urandom(strlen(g_charval));
          memset(&t.val[j], g_charval[v], k);
          j += k;
        }
      }
      break;
    default:
      assert(false);
      break;
    }
    d.ind[i] = 0;
  }
}

static void
makeop(const Op* prev_op, Op* op, Uint32 pk1, Op::Type t)
{
  op->type = t;
  const Data& dp = prev_op->data[0];
  Data& d0 = op->data[0];
  Data& d1 = op->data[1];
  uint i;
  for (i = 0; i < ncol(); i++) {
    const Col& c = getcol(i);
    makedata(c, d0, pk1, t);
    if (t == Op::INS) {
      d1.ind[i] = -1;
    } else if (t == Op::DEL) {
      assert(dp.ind[i] >= 0);
      if (c.pk)
        d1.ind[i] = -1;
      else
        copycol(c, dp, d1);
    } else if (t == Op::UPD) {
      assert(dp.ind[i] >= 0);
      if (d0.ind[i] == -1) // not updating this col
        copycol(c, dp, d0); // must keep track of data
      copycol(c, dp, d1);
    } else {
      assert(false);
    }
  }
  Uint32 pk1_tmp = ~(Uint32)0;
  reqrc(checkop(op, pk1_tmp) == 0);
  reqrc(pk1 == pk1_tmp);
}

static void
makeops()
{
  ll1("makeops");
  Uint32 pk1 = 0;
  while (g_usedops < g_opts.maxops && pk1 < g_opts.maxpk) {
    if (g_opts.opstring == 0)
      pk1 = urandom(g_opts.maxpk);
    ll2("makeops: pk1=" << pk1);
    // total op on the pk so far
    // optype either NUL=initial/deleted or INS=created
    Op* tot_op = g_pk_op[pk1];
    if (tot_op == 0)
      tot_op = g_pk_op[pk1] = getop(Op::OP);
    assert(tot_op->type == Op::NUL || tot_op->type == Op::INS);
    // add new commit chain to end
    Op* last_gci = tot_op;
    while (last_gci->next_gci != 0)
      last_gci = last_gci->next_gci;
    Op* gci_op = getop(Op::OP);
    last_gci->next_gci = gci_op;
    Op* com_op = getop(Op::OP);
    gci_op->next_com = com_op;
    // length of random chain
    uint len = ~0;
    if (g_opts.opstring == 0) {
      len = 1 + urandom(g_maxcom - 1);
      len = 1 + urandom(len - 1); // 2x bias for short chain
    }
    ll2("makeops: com chain");
    uint n = 0;
    while (1) {
      // random or from current g_opts.opstring part
      Op::Type t;
      if (g_opts.opstring == 0) {
        if (n == len)
          break;
        do {
          t = (Op::Type)urandom(g_optypes);
        } while (tot_op->type == Op::NUL && (t == Op::DEL || t == Op::UPD) ||
                 tot_op->type == Op::INS && t == Op::INS);
      } else {
        const char* str = g_opstringpart[g_loop % g_opstringparts];
        uint m = strlen(str);
        uint k = tot_op->num_com + tot_op->num_op;
        assert(k < m);
        char c = str[k];
        if (c == 'c') {
          if (k + 1 == m)
            pk1 += 1;
          break;
        }
        const char* p = "idu";
        const char* q = strchr(p, c);
        assert(q != 0);
        t = (Op::Type)(q - p);
      }
      Op* op = getop(Op::OP);
      makeop(tot_op, op, pk1, t);
      // add to end
      Op* last_op = com_op;
      while (last_op->next_op != 0)
        last_op = last_op->next_op;
      last_op->next_op = op;
      // merge into chain head and total op
      reqrc(compop(com_op, op, com_op) == 0);
      reqrc(compop(tot_op, op, tot_op) == 0);
      assert(tot_op->type == Op::NUL || tot_op->type == Op::INS);
      // counts
      com_op->num_op += 1;
      tot_op->num_op += 1;
      n++;
    }