source: branches/parallel/src/TableTraverse.cpp @ 2283

//
// C++ Interface: Scantable
//
// Description:
//
// Copyright: See COPYING file that comes with this distribution
//

#include <stdio.h>
#include <stdlib.h>
#include <TableTraverse.h>
#include <tables/Tables/TableColumn.h>
#include <tables/Tables/PlainColumn.h>
#include <tables/Tables/DataManager.h>

#define TIMING 0

#if TIMING
#include <sys/time.h>
#endif

static const char version[] = "$Id$";

using namespace casa;

namespace asap {
  class Comparator {
  public:
    virtual ~Comparator() {}
    virtual int compare(const void *a, const void *b) = 0;
  };

  class CompContext: public Comparator {
    uInt nCols;
    void const *const *colValues;
    const TypeManager *const* typeManagers;
  public:
    CompContext(uInt nCols, void const *const colValues[],
                const TypeManager *const typeManagers[])
      : nCols(nCols), colValues(colValues), typeManagers(typeManagers) {
      DebugAssert(colValues != NULL, AipsError);
      DebugAssert(typeManagers != NULL, AipsError);
      for (uInt i = 0; i < nCols; i++) {
        DebugAssert(colValues[i] != NULL, AipsError);
        DebugAssert(typeManagers[i] != NULL, AipsError);
      }
    }

    virtual int compare(const void *a, const void *b) {
      uInt ai = *(uInt*)a; 
      uInt bi = *(uInt*)b; 
      for (size_t i = 0; i < nCols; i++) {
        const size_t valueSize = typeManagers[i]->sizeOf();
        const char *values = (const char *)colValues[i];
        int diff = typeManagers[i]->getComparator()->comp(&values[ai*valueSize], &values[bi*valueSize]);
        if (diff != 0) {
          return diff;
        }
      }
      return 0;
    }
  };
}

#if TIMING
static double currenttime()
{
  struct timeval tv;
  int result = gettimeofday(&tv, NULL);
  return tv.tv_sec + ((double)tv.tv_usec) / 1000000.;
}
#endif

static inline void swap_(char *a, char *b, size_t size)
{
  char tmp[size];
  char *p = tmp;
  do {
    *p = *a;
    *a++ = *b;
    *b++ = *p++;
  } while (--size > 0);
}

static inline void cpy(char *dest, char *src, size_t size)
{
  for (size_t i = 0; i < size; i++) {
    dest[i] = src[i];
  }
}

static void sort_(char *left, char *right, const size_t size,
                  asap::Comparator *comp, int level)
{
  const size_t len = (right - left) / size;
  if (len < 2) {
    return;
  }
  size_t mid = len >> 1;
  char *l = left;
  char *r = right - size;
  char pivot[size]  __attribute__ ((aligned (16)));
  cpy(pivot, &left[mid*size], size);
  while (true) {
    while (l < r && comp->compare(l, pivot) < 0) {
      l += size;
    }
    while (l < r && comp->compare(r, pivot) > 0) {
      r -= size;
        }
    if (l >= r) {
      break;
    }
    swap_(l, r, size);
    l += size;
    r -= size;
  }
  sort_(left, l, size, comp, level + 1);
  sort_(l, right, size, comp, level + 1);
}

static inline void modernQSort(void *base, size_t elements, size_t size,
                               asap::Comparator *comp)
{
  sort_((char *)base, &((char *)base)[elements * size], size, comp, 0);
}

namespace asap {
  class ROTableColumnBackDoor: public ROTableColumn {
  public:
    static BaseColumn *baseColumnPtr(const ROTableColumn *col)
    {
      return ((ROTableColumnBackDoor*)col)->baseColPtr();
    }
  };

  static void copyColumnData(void *colValues, size_t elementSize, uInt nRows,
                             BaseColumn *refCol)
  {
    char *base = (char *)colValues;
    for (uInt i = 0; i < nRows; i++) {
      refCol->get(i, &base[elementSize * i]);
    }
  }

  void traverseTable(const Table &table,
                     const char *const columnNames[],
                     const TypeManager *const typeManagers[],
                     TableVisitor *visitor,
                     Bool doSort)
  {
#if TIMING
    double s = currenttime();
#endif
    uInt colCount = 0;
    for (; columnNames[colCount]; colCount++) {
      AlwaysAssert(typeManagers[colCount], AipsError);
    }

    ROTableColumn *cols[colCount];
    void *colValues[colCount];
    for (uInt i = 0; i < colCount; i++) {
      cols[i] = NULL;
      colValues[i] = NULL;
    }
    size_t sizes[colCount];
    const uInt nRows = table.nrow();
    for (uInt i = 0; i < colCount; i++) {
      cols[i] = new ROTableColumn(table, columnNames[i]);
      colValues[i] = typeManagers[i]->allocArray(nRows);
      sizes[i] = typeManagers[i]->sizeOf();
      BaseColumn *baseCol = ROTableColumnBackDoor::baseColumnPtr(cols[i]);
      PlainColumn *col = dynamic_cast <PlainColumn *>(baseCol);
      if (col) {
        const uInt gotRows = col->dataManagerColumn()->
          getBlockV(0, nRows, colValues[i]);
        DebugAssert(gotRows == nRows, AipsError);
      } else {
        copyColumnData(colValues[i], typeManagers[i]->sizeOf(), nRows, baseCol);
      }
    }

    uInt *idx = new uInt[nRows];
    for (uInt i = 0; i < nRows; i++) {
      idx[i] = i;
    }

    try {
      if (doSort) {
        CompContext compCtx(colCount, colValues, typeManagers);
        modernQSort(idx, nRows, sizeof(idx[0]), &compCtx);
      }

      visitor->start();
      Bool first = True;
      for (uInt i = 0; i < nRows; i++) {
        if (visitor->visit(first, idx[i], colCount, colValues) == False) {
          break;
        }
        first = False;
      }
      visitor->finish();
    } catch (...) {
      delete[] idx;
      for (uInt i = 0; i < colCount; i++) {
        typeManagers[i]->freeArray(colValues[i]);
        delete cols[i];
      }
      throw;
    }

    delete[] idx;
    for (uInt i = 0; i < colCount; i++) {
      typeManagers[i]->freeArray(colValues[i]);
      delete cols[i];
    }

#if TIMING
    double e = currenttime();
    printf("%s took %.3f sec\n", __PRETTY_FUNCTION__, e - s);
#endif
  }
}

#if 0
using namespace asap ;

class MyComp: public Comparator {
public:
  virtual int compare(const void *a, const void *b) {
    return (*(int*)a) - (*(int*)b);
  }
};

int main()
{
  srand((int)currenttime());
  int x[100];
  printf("init\n");
  for (int i = 0; i < 100; i++) {
    x[i] = rand() % 100;
  }
  MyComp myComp;
  printf("sorting\n");
  modernQSort(x, 1, sizeof(x[0]), &myComp);
  modernQSort(x, 2, sizeof(x[0]), &myComp);
  modernQSort(x, 3, sizeof(x[0]), &myComp);
  modernQSort(x, 100, sizeof(x[0]), &myComp);
  for (int i = 0; i < 100; i++) {
    printf("%d\n", x[i]);
  }
}
#endif