src/lib/gscphftaaux/hfta_udaf.cc

   1 /* ------------------------------------------------
   2 Copyright 2014 AT&T Intellectual Property
   3    Licensed under the Apache License, Version 2.0 (the "License");
   4    you may not use this file except in compliance with the License.
   5    You may obtain a copy of the License at
   6
   7      http://www.apache.org/licenses/LICENSE-2.0
   8
   9    Unless required by applicable law or agreed to in writing, software
  10    distributed under the License is distributed on an "AS IS" BASIS,
  11    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12    See the License for the specific language governing permissions and
  13    limitations under the License.
  14  ------------------------------------------- */
  15
  16
  17 #include "gsconfig.h"
  18 #include "gstypes.h"
  19 #include "hfta_udaf.h"
  20 #include "rts_udaf.h"
  21 #include <stdio.h>
  22 #include <limits.h>
  23 #include <math.h>
  24 //#include <memory.h>
  25 #include <string.h>
  26 #include <sys/time.h>
  27 #include <iostream>
  28
  29 #include "hfta_runtime_library.h"
  30
  31
  32 #define max(a,b) ((a) > (b) ? (a) : (b))
  33 #define min(x,y) ((x) < (y) ? (x) : (y))
  34 #define lg(x)    (log(x) / log(2))
  35
  36 using namespace std;
  37
  38
  39 // -------------------------------------------------------------------
  40 //              moving sum over N intervals
  41
  42 struct moving_sum_udaf_str{
  43         gs_uint32_t N;
  44         gs_uint32_t pos;
  45         gs_uint32_t *sums;
  46 };
  47
  48 void moving_sum_udaf_HFTA_AGGR_INIT_(gs_sp_t buf){
  49   struct moving_sum_udaf_str * u = (struct moving_sum_udaf_str *) buf;
  50   u->N=0; u->pos=0; u->sums=NULL;
  51 }
  52
  53 void moving_sum_udaf_HFTA_AGGR_UPDATE_(gs_sp_t buf, gs_uint32_t s, gs_uint32_t N) {
  54   struct moving_sum_udaf_str * u = (struct moving_sum_udaf_str *) buf;
  55   if(u->sums == NULL){
  56         u->sums = (gs_uint32_t *)malloc(N*sizeof(gs_uint32_t));
  57         for(gs_int32_t i=0;i<N;i++)
  58                 u->sums[i] = 0;
  59     u->N = N;
  60   }
  61   u->sums[u->pos] += s;
  62 }
  63
  64 void super_moving_sum_udaf_HFTA_AGGR_UPDATE_(gs_sp_t  buf, gs_uint64_t sub_sum) {
  65   struct moving_sum_udaf_str * u = (struct moving_sum_udaf_str *) buf;
  66   gs_uint32_t s = (gs_uint32_t)(sub_sum & 0xffffffff);
  67   if(u->sums == NULL){
  68     gs_uint32_t N = (gs_uint32_t)((sub_sum & 0xffffffff00000000ull) >> 32);
  69         u->sums = (gs_uint32_t *)malloc(N*sizeof(gs_uint32_t));
  70         for(gs_int32_t i=0;i<N;i++)
  71                 u->sums[i] = 0;
  72     u->N = N;
  73   }
  74   u->sums[u->pos] += s;
  75 }
  76
  77 void moving_sum_udaf_HFTA_AGGR_OUTPUT_(gs_p_t *result, gs_sp_t  buf){
  78         *result = (gs_p_t)(buf);
  79 }
  80
  81 void moving_sum_udaf_HFTA_AGGR_DESTROY_(gs_sp_t  buf){
  82   struct moving_sum_udaf_str * u = (struct moving_sum_udaf_str *) buf;
  83   if(u->sums != NULL)
  84         free(u->sums);
  85 }
  86
  87 void moving_sum_udaf_HFTA_AGGR_REINIT_( gs_sp_t  buf){
  88   struct moving_sum_udaf_str * u = (struct moving_sum_udaf_str *) buf;
  89   u->pos++;
  90   if(u->pos >= u->N)
  91         u->pos = 0;
  92   u->sums[u->pos] = 0;
  93 }
  94
  95 gs_uint32_t moving_sum_extract(gs_p_t result){
  96   struct moving_sum_udaf_str * u = (struct moving_sum_udaf_str *) result;
  97   gs_uint32_t s=0, i=0;
  98   for(i=0; i<u->N;i++){
  99     s += u->sums[i];
 100   }
 101   return s;
 102 }
 103
 104 gs_float_t moving_sum_extract_exp(gs_p_t result, gs_float_t alpha){
 105   struct moving_sum_udaf_str * u = (struct moving_sum_udaf_str *) result;
 106   gs_uint32_t p=0, i=0;
 107   gs_float_t s=0.0, m=1.0;
 108   p=u->pos;
 109   for(i=0; i<u->N;i++){
 110     s += u->sums[i]*m;
 111     if(p==0)
 112                 p=u->N - 1;
 113     else
 114                 p--;
 115         m *= alpha;
 116   }
 117   return s;
 118 }
 119
 120
 121 // -------------------------------------------------------------------
 122 //              sum over 3 intervals : test rUDAF
 123
 124 struct sum3_udaf_str{
 125   gs_uint32_t s_2;
 126   gs_uint32_t s_1;
 127   gs_uint32_t s_0;
 128 };
 129
 130 void sum3_HFTA_AGGR_INIT_(gs_sp_t  buf) {
 131   struct sum3_udaf_str * u = (struct sum3_udaf_str *) buf;
 132   u->s_0 = 0; u->s_1 = 0; u->s_2 = 0;
 133   return;
 134 }
 135
 136 void sum3_HFTA_AGGR_UPDATE_(gs_sp_t  buf, gs_uint32_t s) {
 137   struct sum3_udaf_str * u = (struct sum3_udaf_str *) buf;
 138   u->s_0 += s;
 139   return;
 140 }
 141
 142 void sum3_HFTA_AGGR_OUTPUT_(gs_uint32_t *result, gs_sp_t  buf) {
 143   struct sum3_udaf_str * u = (struct sum3_udaf_str *) buf;
 144   *result = u->s_0 + u->s_1 + u->s_2;
 145   return;
 146 }
 147
 148 void sum3_HFTA_AGGR_DESTROY_(gs_sp_t  buf) {
 149   return;
 150 }
 151
 152 void sum3_HFTA_AGGR_REINIT_( gs_sp_t  buf) {
 153   struct sum3_udaf_str * u = (struct sum3_udaf_str *) buf;
 154   u->s_2 = u->s_1;
 155   u->s_1 = u->s_0;
 156   u->s_0 = 0;
 157   return;
 158 }
 159
 160
 161 #define HISTORY_LENGTH 1024
 162
 163 /////////////////////////////////////////////////////////////////////////
 164 /////   Calculate the average of all positive gs_float_t numbers
 165
 166 struct posavg_struct{
 167   gs_float_t sum;
 168   gs_float_t cnt;
 169 };
 170
 171 void POSAVG_HFTA_AGGR_INIT_(gs_sp_t  buf) {
 172   struct posavg_struct * a = (struct posavg_struct *) buf;
 173   a->sum=0;
 174   a->cnt=0;
 175   return;
 176 }
 177
 178 void POSAVG_HFTA_AGGR_UPDATE_(gs_sp_t  buf, gs_float_t v) {
 179   struct posavg_struct * a = (struct posavg_struct *) buf;
 180   if (v>=0) {
 181     a->sum=a->sum+v;
 182     a->cnt=a->cnt+1;
 183   }
 184   return;
 185 }
 186
 187 void POSAVG_HFTA_AGGR_OUTPUT_(gs_float_t * v, gs_sp_t  buf) {
 188   struct posavg_struct * a = (struct posavg_struct *) buf;
 189   if (a->cnt>0) {
 190     *v=(a->sum/a->cnt);
 191   } else {
 192     *v=-1;
 193   }
 194   return;
 195 }
 196
 197 void POSAVG_HFTA_AGGR_DESTROY_(gs_sp_t  buf) {
 198   return;
 199 }
 200
 201 /////////////////////////////////////////////////////////////////////////
 202 /////                   avg_udaf (simple example)
 203
 204 //              struct received from subaggregate
 205 struct avg_udaf_lfta_struct_t{
 206         gs_int64_t  sum;
 207         gs_uint32_t cnt;
 208 };
 209
 210 //              sctarchpad struct
 211 struct avg_udaf_hfta_struct_t{
 212         gs_int64_t sum;
 213         gs_uint32_t cnt;
 214 };
 215
 216 //                      avg_udaf functions
 217 void avg_udaf_HFTA_AGGR_INIT_(gs_sp_t b){
 218         avg_udaf_hfta_struct_t *s = (avg_udaf_hfta_struct_t *) b;
 219         s->sum = 0;
 220         s->cnt = 0;
 221 }
 222
 223 void avg_udaf_HFTA_AGGR_UPDATE_(gs_sp_t b, gs_uint32_t v){
 224         avg_udaf_hfta_struct_t *s = (avg_udaf_hfta_struct_t *) b;
 225         s->sum += v;
 226         s->cnt ++;
 227 }
 228
 229 void avg_udaf_HFTA_AGGR_OUTPUT_(vstring *r,gs_sp_t b){
 230         r->length = 12;
 231         r->offset = (gs_p_t)(b);
 232         r->reserved = SHALLOW_COPY;
 233 }
 234
 235 void avg_udaf_HFTA_AGGR_DESTROY_(gs_sp_t b){
 236         return;
 237 }
 238
 239
 240 //                      avg_udaf superaggregate functions
 241 void avg_udaf_hfta_HFTA_AGGR_INIT_(gs_sp_t b){
 242         avg_udaf_hfta_struct_t *s = (avg_udaf_hfta_struct_t *) b;
 243         s->sum = 0;
 244         s->cnt = 0;
 245 }
 246
 247 void avg_udaf_hfta_HFTA_AGGR_UPDATE_(gs_sp_t b, vstring *v){
 248         if(v->length != 12) return;
 249         avg_udaf_hfta_struct_t *s = (avg_udaf_hfta_struct_t *) b;
 250         avg_udaf_lfta_struct_t *vs = (avg_udaf_lfta_struct_t *)(v->offset);
 251         s->sum += vs->sum;
 252         s->cnt += vs->cnt;
 253 }
 254
 255 void avg_udaf_hfta_HFTA_AGGR_OUTPUT_(vstring *r,gs_sp_t b){
 256         r->length = 12;
 257         r->offset = (gs_p_t)(b);
 258         r->reserved = SHALLOW_COPY;
 259 }
 260
 261 void avg_udaf_hfta_HFTA_AGGR_DESTROY_(gs_sp_t b){
 262         return;
 263 }
 264
 265 //              Extraction function
 266 gs_float_t extr_avg_fcn(vstring *v){
 267         if(v->length != 12) return 0;
 268         avg_udaf_hfta_struct_t *vs = (avg_udaf_hfta_struct_t *)(v->offset);
 269         gs_float_t r = (gs_float_t)(vs->sum) / vs->cnt;
 270     return r;
 271 }
 272
 273 /////////////////////////////////////////////////////////
 274 //              FIRST aggregate
 275 // hfta only
 276
 277 void FIRST_HFTA_AGGR_INIT_(gs_uint32_t* scratch) {
 278         *scratch = UINT_MAX;            // we will encode uninitialized value of UINT_MAX
 279 }
 280 void FIRST_HFTA_AGGR_REINIT_(gs_uint32_t* scratch) { }
 281 void FIRST_HFTA_AGGR_UPDATE_(gs_uint32_t* scratch, gs_uint32_t val) {
 282         if (*scratch == UINT_MAX)
 283                 *scratch = val;
 284 }
 285 void FIRST_HFTA_AGGR_OUTPUT_(gs_uint32_t* res, gs_uint32_t* scratch) {
 286         *res = *scratch;
 287 }
 288 void FIRST_HFTA_AGGR_DESTROY_(gs_uint32_t* scratch) { }
 289
 290 void FIRST_HFTA_AGGR_INIT_(gs_uint64_t* scratch) {
 291         *scratch = UINT_MAX;            // we will encode uninitialized value of UINT_MAX
 292 }
 293 void FIRST_HFTA_AGGR_REINIT_(gs_uint64_t* scratch) { }
 294 void FIRST_HFTA_AGGR_UPDATE_(gs_uint64_t* scratch, gs_uint64_t val) {
 295         if (*scratch == UINT_MAX)
 296                 *scratch = val;
 297 }
 298 void FIRST_HFTA_AGGR_OUTPUT_(gs_uint64_t* res, gs_uint64_t* scratch) {
 299         *res = *scratch;
 300 }
 301 void FIRST_HFTA_AGGR_DESTROY_(gs_uint64_t* scratch) { }
 302
 303
 304
 305 void FIRST_ULL_HFTA_AGGR_INIT_(gs_uint64_t* scratch) {
 306         *scratch = UINT_MAX;            // we will encode uninitialized value of UINT_MAX
 307 }
 308
 309 void FIRST_ULL_HFTA_AGGR_REINIT_(gs_uint64_t* scratch) { }
 310
 311 void FIRST_ULL_HFTA_AGGR_UPDATE_(gs_uint64_t* scratch, gs_uint64_t val) {
 312         if (*scratch == UINT_MAX)
 313                 *scratch = val;
 314 }
 315
 316 void FIRST_ULL_HFTA_AGGR_OUTPUT_(gs_uint64_t* res, gs_uint64_t* scratch) {
 317         *res = *scratch;
 318 }
 319
 320 void FIRST_ULL_HFTA_AGGR_DESTROY_(gs_uint64_t* scratch) { }
 321
 322
 323 void FIRST_STR_HFTA_AGGR_INIT_(vstring* scratch) {
 324         scratch->offset= 0;
 325 }
 326
 327 void FIRST_STR_HFTA_AGGR_REINIT_(vstring* scratch) { }
 328
 329 void FIRST_STR_HFTA_AGGR_UPDATE_(vstring* scratch, vstring* val) {
 330         if (!scratch->offset) {
 331     scratch->length = val->length;
 332     scratch->offset = val->offset;
 333     scratch->reserved = SHALLOW_COPY;
 334         }
 335 }
 336
 337 void FIRST_STR_HFTA_AGGR_OUTPUT_(vstring* res, vstring* scratch) {
 338         *res = *scratch;
 339 }
 340
 341 void FIRST_STR_HFTA_AGGR_DESTROY_(vstring* scratch) { }
 342
 343 // hfta/lfta split
 344
 345 void FIRST_hfta_HFTA_AGGR_INIT_(gs_uint32_t* scratch) {
 346         *scratch = UINT_MAX;            // we will encode uninitialized value of UINT_MAX
 347 }
 348
 349 void FIRST_hfta_HFTA_AGGR_REINIT_(gs_uint32_t* scratch) { }
 350
 351 void FIRST_hfta_HFTA_AGGR_UPDATE_(gs_uint32_t* scratch, gs_uint32_t val) {
 352         if (*scratch == UINT_MAX)
 353                 *scratch = val;
 354 }
 355
 356 void FIRST_hfta_HFTA_AGGR_OUTPUT_(gs_uint32_t* res, gs_uint32_t* scratch) {
 357         *res = *scratch;
 358 }
 359
 360 void FIRST_hfta_HFTA_AGGR_DESTROY_(gs_uint32_t* scratch) { }
 361
 362 void FIRST_ULL_hfta_HFTA_AGGR_INIT_(gs_uint64_t* scratch) {
 363         *scratch = UINT_MAX;            // we will encode uninitialized value of UINT_MAX
 364 }
 365
 366 void FIRST_ULL_hfta_HFTA_AGGR_REINIT_(gs_uint64_t* scratch) { }
 367
 368 void FIRST_ULL_hfta_HFTA_AGGR_UPDATE_(gs_uint64_t* scratch, gs_uint64_t val) {
 369         if (*scratch == UINT_MAX)
 370                 *scratch = val;
 371 }
 372
 373 void FIRST_ULL_hfta_HFTA_AGGR_OUTPUT_(gs_uint64_t* res, gs_uint64_t* scratch) {
 374         *res = *scratch;
 375 }
 376
 377 void FIRST_ULL_hfta_HFTA_AGGR_DESTROY_(gs_uint64_t* scratch) { }
 378
 379
 380 void FIRST_STR_hfta_HFTA_AGGR_INIT_(vstring* scratch) {
 381         scratch->offset= 0;
 382 }
 383
 384 void FIRST_STR_hfta_HFTA_AGGR_REINIT_(vstring* scratch) { }
 385
 386 void FIRST_STR_hfta_HFTA_AGGR_UPDATE_(vstring* scratch, vstring* val) {
 387         if (!scratch->offset) {
 388     scratch->length = val->length;
 389     scratch->offset = val->offset;
 390     scratch->reserved = SHALLOW_COPY;
 391         }
 392 }
 393
 394 void FIRST_STR_hfta_HFTA_AGGR_OUTPUT_(vstring* res, vstring* scratch) {
 395         *res = *scratch;
 396 }
 397
 398 void FIRST_STR_hfta_HFTA_AGGR_DESTROY_(vstring* scratch) { }
 399
 400
 401 /////////////////////////////////////////////////////////
 402 //              LAST aggregate
 403
 404 // hfta only
 405
 406 void LAST_HFTA_AGGR_INIT_(gs_uint32_t* scratch) { }
 407 void LAST_HFTA_AGGR_REINIT_(gs_uint32_t* scratch) { }
 408 void LAST_HFTA_AGGR_UPDATE_(gs_uint32_t* scratch, gs_uint32_t val) {
 409         *scratch = val;
 410 }
 411 void LAST_HFTA_AGGR_OUTPUT_(gs_uint32_t* res, gs_uint32_t* scratch) {
 412         *res = *scratch;
 413 }
 414 void LAST_HFTA_AGGR_DESTROY_(gs_uint32_t* scratch) { }
 415
 416
 417 void LAST_HFTA_AGGR_INIT_(gs_uint64_t* scratch) { }
 418 void LAST_HFTA_AGGR_REINIT_(gs_uint64_t* scratch) { }
 419 void LAST_HFTA_AGGR_UPDATE_(gs_uint64_t* scratch, gs_uint64_t val) {
 420         *scratch = val;
 421 }
 422 void LAST_HFTA_AGGR_OUTPUT_(gs_uint64_t* res, gs_uint64_t* scratch) {
 423         *res = *scratch;
 424 }
 425 void LAST_HFTA_AGGR_DESTROY_(gs_uint64_t* scratch) { }
 426
 427
 428
 429 void LAST_ULLHFTA_AGGR_INIT_(gs_uint64_t* scratch) { }
 430
 431 void LAST_ULL_HFTA_AGGR_REINIT_(gs_uint64_t* scratch) { }
 432
 433 void LAST_ULL_HFTA_AGGR_UPDATE_(gs_uint64_t* scratch, gs_uint64_t val) {
 434         *scratch = val;
 435 }
 436
 437 void LAST_ULL_HFTA_AGGR_OUTPUT_(gs_uint64_t* res, gs_uint64_t* scratch) {
 438         *res = *scratch;
 439 }
 440
 441 void LAST_ULL_HFTA_AGGR_DESTROY_(gs_uint64_t* scratch) { }
 442
 443
 444 void LAST_STR_HFTA_AGGR_INIT_(vstring* scratch) {
 445         scratch->offset= 0;
 446 }
 447
 448 void LAST_STR_HFTA_AGGR_REINIT_(vstring* scratch) { }
 449
 450 void LAST_STR_HFTA_AGGR_UPDATE_(vstring* scratch, vstring* val) {
 451         scratch->length = val->length;
 452   scratch->offset = val->offset;
 453   scratch->reserved = SHALLOW_COPY;
 454 }
 455
 456 void LAST_STR_HFTA_AGGR_OUTPUT_(vstring* res, vstring* scratch) {
 457         *res = *scratch;
 458 }
 459
 460 void LAST_STR_HFTA_AGGR_DESTROY_(vstring* scratch) { }
 461
 462 // hfta/lfta split
 463
 464 void LAST_hfta_HFTA_AGGR_INIT_(gs_uint32_t* scratch) { }
 465
 466 void LAST_hfta_HFTA_AGGR_REINIT_(gs_uint32_t* scratch) { }
 467
 468 void LAST_hfta_HFTA_AGGR_UPDATE_(gs_uint32_t* scratch, gs_uint32_t val) {
 469         *scratch = val;
 470 }
 471
 472 void LAST_hfta_HFTA_AGGR_OUTPUT_(gs_uint32_t* res, gs_uint32_t* scratch) {
 473         *res = *scratch;
 474 }
 475
 476 void LAST_hfta_HFTA_AGGR_DESTROY_(gs_uint32_t* scratch) { }
 477
 478 void LAST_ULL_hfta_HFTA_AGGR_INIT_(gs_uint64_t* scratch) { }
 479
 480 void LAST_ULL_hfta_HFTA_AGGR_REINIT_(gs_uint64_t* scratch) { }
 481
 482 void LAST_ULL_hfta_HFTA_AGGR_UPDATE_(gs_uint64_t* scratch, gs_uint64_t val) {
 483         *scratch = val;
 484 }
 485
 486 void LAST_ULL_hfta_HFTA_AGGR_OUTPUT_(gs_uint64_t* res, gs_uint64_t* scratch) {
 487         *res = *scratch;
 488 }
 489
 490 void LAST_ULL_hfta_HFTA_AGGR_DESTROY_(gs_uint64_t* scratch) { }
 491
 492
 493 void LAST_STR_hfta_HFTA_AGGR_INIT_(vstring* scratch) {
 494         scratch->offset= 0;
 495 }
 496
 497 void LAST_STR_hfta_HFTA_AGGR_REINIT_(vstring* scratch) { }
 498
 499 void LAST_STR_hfta_HFTA_AGGR_UPDATE_(vstring* scratch, vstring* val) {
 500         scratch->length = val->length;
 501   scratch->offset = val->offset;
 502   scratch->reserved = SHALLOW_COPY;
 503 }
 504
 505 void LAST_STR_hfta_HFTA_AGGR_OUTPUT_(vstring* res, vstring* scratch) {
 506         *res = *scratch;
 507 }
 508
 509 void LAST_STR_hfta_HFTA_AGGR_DESTROY_(vstring* scratch) { }
 510
 511
 512 /////////////////////////////////////////////////////////
 513 //              running_array_aggr aggregate
 514
 515 void running_array_aggr_hfta_HFTA_AGGR_INIT_(vstring* scratch) {
 516   scratch->offset = NULL;
 517   scratch->length = 0;
 518 }
 519
 520 void running_array_aggr_hfta_HFTA_AGGR_REINIT_(vstring* scratch) { }
 521
 522 void running_array_aggr_hfta_HFTA_AGGR_UPDATE_(vstring* scratch, vstring* val) {
 523   char buffer[100];
 524
 525   gs_uint32_t* ints = (gs_uint32_t*)val->offset;
 526   switch (val->length / sizeof (gs_uint32_t)) {
 527     case 4:
 528       sprintf(buffer, "%u,%u,%u,%u", ints[0], ints[1], ints[2], ints[3]);
 529       break;
 530     case 3:
 531       sprintf(buffer, "%u,%u,%u", ints[0], ints[1], ints[2]);
 532       break;
 533     case 2:
 534       sprintf(buffer, "%u,%u", ints[0], ints[1]);
 535       break;
 536     case 1:
 537       sprintf(buffer, "%u", ints[0]);
 538       break;
 539     case 0:
 540       return;
 541   }
 542   int buf_len = strlen(buffer);
 543
 544   // append the content of buffer to scratch
 545         if (!scratch->offset) {
 546     Vstring_Constructor(scratch, buffer);
 547         } else {
 548     scratch->offset = (gs_p_t)realloc((void*)scratch->offset, scratch->length + buf_len + 1);
 549     *((char*)scratch->offset + scratch->length) = ',';
 550     memcpy((void*)(scratch->offset + scratch->length + 1), (void*)buffer, buf_len);
 551     scratch->length += buf_len + 1;
 552     scratch->reserved = INTERNAL;
 553   }
 554 }
 555
 556 void running_array_aggr_hfta_HFTA_AGGR_OUTPUT_(vstring* res, vstring* scratch) {
 557         *res = *scratch;
 558   res->reserved = SHALLOW_COPY;
 559 }
 560
 561 void running_array_aggr_hfta_HFTA_AGGR_DESTROY_(vstring* scratch) {
 562   hfta_vstr_destroy(scratch);
 563  }
 564