GRASS GIS 8 Programmer's Manual  8.5.0dev(2025)-eb16a2cfc9
All Data Structures Files Functions Variables Typedefs Enumerations Enumerator Friends Macros Pages
embuffer.h
Go to the documentation of this file.
1 /****************************************************************************
2  *
3  * MODULE: iostream
4  *
5 
6  * COPYRIGHT (C) 2007 Laura Toma
7  *
8  *
9 
10  * Iostream is a library that implements streams, external memory
11  * sorting on streams, and an external memory priority queue on
12  * streams. These are the fundamental components used in external
13  * memory algorithms.
14 
15  * Credits: The library was developed by Laura Toma. The kernel of
16  * class STREAM is based on the similar class existent in the GPL TPIE
17  * project developed at Duke University. The sorting and priority
18  * queue have been developed by Laura Toma based on communications
19  * with Rajiv Wickremesinghe. The library was developed as part of
20  * porting Terraflow to GRASS in 2001. PEARL upgrades in 2003 by
21  * Rajiv Wickremesinghe as part of the Terracost project.
22 
23  *
24  * This program is free software; you can redistribute it and/or modify
25  * it under the terms of the GNU General Public License as published by
26  * the Free Software Foundation; either version 2 of the License, or
27  * (at your option) any later version.
28  *
29 
30  * This program is distributed in the hope that it will be useful,
31  * but WITHOUT ANY WARRANTY; without even the implied warranty of
32  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
33  * General Public License for more details. *
34  * **************************************************************************/
35 
36 #ifndef __EMBUFFER_H
37 #define __EMBUFFER_H
38 
39 #include <stdio.h>
40 #include <assert.h>
41 #include <stdlib.h>
42 #include <math.h>
43 
44 #include "ami_config.h" //for SAVE_MEMORY
45 #include "ami_stream.h"
46 #include "mm.h"
47 #include "mm_utils.h"
48 #include "pqheap.h"
49 
50 #define MY_LOG_DEBUG_ID(x) // inhibit debug printing
51 // #define MY_LOG_DEBUG_ID(x) LOG_DEBUG_ID(x)
52 
53 /**********************************************************
54  DEBUGGING FLAGS
55 ***********************************************************/
56 
57 // setting this enables checking that the streams/arrays inserted in
58 // buffers are sorted in increasing order
59 // #define EMBUF_CHECK_INSERT
60 
61 // enable checking that stream name is the same as the one stored in
62 // the buffer name[]
63 // #define EMBUF_CHECK_NAME
64 
65 // enable printing names as they are checked
66 // #define EMBUF_CHECK_NAME_PRINT
67 
68 // enable printing when streams in a buffer are shifted left to
69 // check that names are shifted accordingly
70 // #define EMBUF_DELETE_STREAM_PRINT
71 
72 // enable printing the name of the stream which is inserted in buff
73 // #define EMBUF_PRINT_INSERT
74 
75 // enable printing the stream names/sizes in cleanup()
76 // #define EMBUF_CLEANUP_PRINT
77 
78 // enable printing when get/put_stream is called (for each stream)
79 // #define EMBUF_PRINT_GETPUT_STREAM
80 
81 // enable printing when get/put_streams is called
82 // #define EMBUF_PRINT_GETPUT_STREAMS
83 
84 /***********************************************************/
85 
86 /*****************************************************************/
87 /* encapsulation of the key together with stream_id; used during
88  stream merging to save space;
89 */
90 template <class KEY>
91 class merge_key {
92 public:
93  KEY k;
94  unsigned int str_id; // id of the stream where key comes from
95 
96 public:
97  merge_key() : str_id(0) {}
98 
99  merge_key(const KEY &x, const unsigned int sid) : k(x), str_id(sid) {}
100 
102 
103  void set(const KEY &x, const unsigned int sid)
104  {
105  k = x;
106  str_id = sid;
107  }
108  KEY key() const { return k; }
109  unsigned int stream_id() const { return str_id; }
110  KEY getPriority() const { return k; }
111 
112  friend ostream &operator<<(ostream &s, const merge_key<KEY> &x)
113  {
114  return s << "<str_id=" << x.str_id << "> " << x.k << " ";
115  }
116  friend int operator<(const merge_key &x, const merge_key &y)
117  {
118  return (x.k < y.k);
119  }
120  friend int operator<=(const merge_key &x, const merge_key &y)
121  {
122  return (x.k <= y.k);
123  }
124  friend int operator>(const merge_key &x, const merge_key &y)
125  {
126  return (x.k > y.k);
127  }
128  friend int operator>=(const merge_key &x, const merge_key &y)
129  {
130  return (x.k >= y.k);
131  }
132  friend int operator!=(const merge_key &x, const merge_key &y)
133  {
134  return (x.k != y.k);
135  }
136  friend int operator==(const merge_key &x, const merge_key &y)
137  {
138  return (x.k == y.k);
139  }
140  friend merge_key operator+(const merge_key &x, const merge_key &y UNUSED)
141  {
142  assert(0);
143  return x;
144  // Key sum = x.k + y.k;
145  // merge_key f(sum, x.str_id);
146  // return f;
147  }
148 };
149 
150 /*****************************************************************
151  *****************************************************************
152  *****************************************************************
153 
154  external_memory buffer
155 
156  Each level-i buffer can store up to <arity>^i * <basesize> items,
157  where typically <arity> is \theta(m) and <basesize> is \theta(M);
158  therefore log_m{n/m} buffers are needed to store N items, one
159  buffer for each level 1..log_m{n/m}. All buffers must have same
160  values or <arity> and <basesize>.
161 
162  Functionality:
163 
164  A level-i on-disk buffer stores <arity>^i * <basesize> items of
165  data, organized in <arity> streams of <arity>^{i-1} items each;
166  <basesize> is same for all buffers and equal to the size of the
167  level 0 buffer (in memory buffer).
168 
169  Invariant: all the <arity> streams of a level-i buffer are in
170  sorted order; in this way sorting the buffer is done by merging the
171  <arity> streams in linear time.
172 
173  Items are inserted in level i-buffer only a whole stream at a time
174  (<arity>^{i-1}*<basesize> items). When all the <arity> streams of
175  the buffer are full, the buffer is sorted and emptied into a stream
176  of a level (i+1)-buffer.
177 
178  The <arity> streams of a buffer are allocated contigously from left
179  to r ight. The unused streams are NULL; The buffer keeps the index of
180  the last used(non-NULL) stream. When a buffer becomes full and is
181  empty, all its buffers are set to NULL.
182 
183  *****************************************************************
184  *****************************************************************
185  ***************************************************************** */
186 
187 /* T is a type with priority of type K and method getPriority() */
188 template <class T, class Key>
189 class em_buffer {
190 private:
191  // number of streams in a buffer;
192  unsigned int arity;
193 
194  // level of buffer: between 1 and log_arity{n/arity}; (level-0 buffer
195  // has a slightly different behaviour so it is implemented as a
196  // different class <im_buffer>)
197  unsigned short level;
198 
199  // level-i buffer contains m streams of data, each of size
200  // arity^{i-1}*basesize;
201  AMI_STREAM<T> **data;
202 
203  // the buffers can be depleted to fill the internal pq;
204  // keep an array which counts, for each stream, how many elements
205  // have been deleted (implicitly from the beginning of stream)
206  long *deleted;
207 
208  // nb of items in each substream; this can be found out by calling
209  // stream_len() on the stream, but it is more costly esp in the case
210  // when streams are on disk and must be moved in and out just to find
211  // stream length; streamsize is set only at stream creation, and the
212  // actual size must subtract the number of iteme deleted from the
213  // bos
214  unsigned long *streamsize;
215 
216  // index of the next available(empty) stream (out of the total m
217  // streams in the buffer);
218  unsigned int index;
219 
220  // nb of items in a stream of level_1 buffer
221  unsigned long basesize;
222 
223 public:
224  // create a level-i buffer of given basesize;
225  em_buffer(const unsigned short i, const unsigned long bs,
226  const unsigned int ar);
227 
228  // copy constructor;
229  em_buffer(const em_buffer &buf);
230 
231  // free the stream array and the streams pointers
232  ~em_buffer();
233 
234  // return the level of the buffer;
235  unsigned short get_level() const { return level; }
236 
237  // return the ith stream (load stream in memory)
238  AMI_STREAM<T> *get_stream(unsigned int i);
239 
240  // return a pointer to the streams of the buffer (loads streams in
241  // memory)
243 
244  // put the ith stream back to disk
245  void put_stream(unsigned int i);
246 
247  // called in pair with get_streams to put all streams back to disk
248  void put_streams();
249 
250  // return a pointer to the array of deletion count for each stream
251  long *get_bos() const { return deleted; }
252 
253  // return the index of the last stream in buffer which contains data;
254  unsigned int laststream() const { return index - 1; }
255 
256  // return the index of the next available stream in the buffer
257  unsigned int nextstream() const { return index; }
258 
259  // increment the index of the next available stream in the buffer
260  void incr_nextstream() { ++index; }
261 
262  // return nb of (non-empty) streams in buffer
263  unsigned int get_nbstreams() const { return index; }
264 
265  // return arity
266  unsigned int get_arity() const { return arity; }
267 
268  // return total nb of deleted elements in all active streams of the buffer
269  long total_deleted() const
270  {
271  long tot = 0;
272  for (unsigned int i = 0; i < index; i++) {
273  tot += deleted[i];
274  }
275  return tot;
276  }
277 
278  // mark as deleted one more element from i'th stream
279  void incr_deleted(unsigned int i)
280  {
281  assert(i < index);
282  deleted[i]++;
283  }
284 
285  // return the nominal size of a stream (nb of items):
286  // arity^{level-1}*basesize;
287  unsigned long get_stream_maxlen() const
288  {
289  return (unsigned long)pow((double)arity, (double)level - 1) * basesize;
290  }
291 
292  // return the actual size of stream i; i must be the index of a valid
293  // stream
294  unsigned long get_stream_len(unsigned int i)
295  {
296  // assert(i>= 0 && i<index);
297  return streamsize[i] - deleted[i];
298  }
299 
300  // return the total current size of the buffer; account for the
301  // deleted elements;
302  unsigned long get_buf_len()
303  {
304  unsigned long tot = 0;
305  for (unsigned int i = 0; i < index; i++) {
306  tot += get_stream_len(i);
307  }
308  return tot;
309  }
310 
311  // return the total maximal capacity of the buffer
312  unsigned long get_buf_maxlen() { return arity * get_stream_maxlen(); }
313 
314  // return true if buffer is empty (all streams are empty)
315  bool is_empty() { return ((nextstream() == 0) || (get_buf_len() == 0)); }
316 
317  // return true if buffer is full(all streams are full)
318  bool is_full() const { return (nextstream() == arity); }
319 
320  // reset
321  void reset();
322 
323  // clean buffer: in case some streams have been emptied by deletion
324  // delete them and shift streams left;
325  void cleanup();
326 
327  // create and return a stream which contains all elements of all
328  // streams of the buffer in sorted ascending order of their
329  // keys(priorities);
330  AMI_STREAM<T> *sort();
331 
332  // insert an array into the buffer; can only insert one
333  // level-i-full-stream-len nb of items at a time; assume the length
334  // of the array is precisely the streamlen of level-i buffer n =
335  // (pow(arity,level-1)*basesize); assume array is sorted; return the
336  // number of items actually inserted
337  long insert(T *a, long n);
338 
339  // insert a stream into the buffer; assume the length of the stream
340  // is precisely the streamlen of level-i buffer n =
341  // (pow(arity,level-1)*basesize); the <nextstream> pointer of buffer
342  // is set to point to the argument stream; (in this way no stream
343  // copying is done, just one pointer copy). The user should be aware
344  // the argument stream is 'lost' - that is a stream which cannot be
345  // inserted repeatedly into many buffers because this would lead to
346  // several buffers pointing to the same stream.
347 
348  // stream is assumed sorted; bos = how many elements are deleted
349  // from the beginning of stream;
350 
351  // return the number of items actually inserted
352  long insert(AMI_STREAM<T> *str, long bos = 0);
353 
354  // print range of elements in buffer
355  void print_range();
356 
357  // print all elements in buffer
358  void print();
359 
360  // prints the sizes of the streams in the buffer
361  void print_stream_sizes();
362 
363  // print the elements in the buffer
364  friend ostream &operator<<(ostream &s, em_buffer &b)
365  {
366  s << "BUFFER_" << b.level << ": ";
367  if (b.index == 0) {
368  s << "[]";
369  }
370  s << "\n";
371  b.get_streams();
372  for (unsigned int i = 0; i < b.index; i++) {
373  b.print_stream(s, i);
374  }
375  b.put_streams();
376  return s;
377  }
378 
379 private:
380  // merge the input streams; there are <arity> streams in total;
381  // write output in <outstream>; the input streams are assumed sorted
382  // in increasing order of their keys;
383  AMI_err substream_merge(AMI_STREAM<T> **instreams, unsigned int arity,
384  AMI_STREAM<T> *outstream);
385 
386  // print to stream the elements in i'th stream
387  void print_stream(ostream &s, unsigned int i);
388 
389 #ifdef SAVE_MEMORY
390  // array of names of streams;
391  char **name;
392 
393  // return the designated name for stream i
394  char *get_stream_name(unsigned int i) const;
395 
396  // print all stream names in buffer
397  void print_stream_names();
398 
399  // checks that name[i] is the same as stream name; stream i must be in
400  // memory (by a previous get_stream call, for instance) in order to
401  // find its length
402  void check_name(unsigned int i);
403 #endif
404 };
405 
406 /************************************************************/
407 // create a level-i buffer of given basesize;
408 template <class T, class Key>
409 em_buffer<T, Key>::em_buffer(const unsigned short i, const unsigned long bs,
410  const unsigned int ar)
411  : arity(ar), level(i), basesize(bs)
412 {
413 
414  assert(level >= 1);
415 
416  char str[100];
417  snprintf(str, sizeof(str),
418  "em_buffer: allocate %d AMI_STREAM*, total %ld\n", arity,
419  (long)(arity * sizeof(AMI_STREAM<T> *)));
420  MEMORY_LOG(str);
421  // allocate STREAM* array
422  data = new AMI_STREAM<T> *[arity];
423 
424  // allocate deleted array
425  snprintf(str, sizeof(str), "em_buffer: allocate deleted array: %ld\n",
426  (long)(arity * sizeof(long)));
427  MEMORY_LOG(str);
428  deleted = new long[arity];
429 
430  // allocate streamsize array
431  snprintf(str, sizeof(str), "em_buffer: allocate streamsize array: %ld\n",
432  (long)(arity * sizeof(long)));
433  MEMORY_LOG(str);
434  streamsize = new unsigned long[arity];
435 
436 #ifdef SAVE_MEMORY
437  // allocate name array
438  snprintf(str, sizeof(str), "em_buffer: allocate name array: %ld\n",
439  (long)(arity * sizeof(char *)));
440  MEMORY_LOG(str);
441  name = new char *[arity];
442  assert(name);
443 #endif
444 
445  // assert data
446  if ((!data) || (!deleted) || (!streamsize)) {
447  cerr << "em_buffer: cannot allocate\n";
448  exit(1);
449  }
450 
451  // initialize the <arity> streams to NULL, deleted[], streamsize[]
452  // and name[]
453  for (unsigned int ui = 0; ui < arity; ui++) {
454  data[ui] = NULL;
455  deleted[ui] = 0;
456  streamsize[ui] = 0;
457 #ifdef SAVE_MEMORY
458  name[ui] = NULL;
459 #endif
460  }
461  // set index
462  index = 0;
463 
464 #ifdef SAVE_MEMORY
465  // streams_in_memory = false;
466 #endif
467 }
468 
469 /************************************************************/
470 // copy constructor;
471 template <class T, class Key>
473  : level(buf.level), basesize(buf.basesize), index(buf.index),
474  arity(buf.arity)
475 {
476 
477  assert(0); // should not get called
478 
479  MEMORY_LOG("em_buffer: copy constr start\n");
480  get_streams();
481  for (unsigned int i = 0; i < index; i++) {
482  assert(data[i]);
483  delete data[i]; // delete old stream if existing
484  data[i] = NULL;
485 
486  // call copy constructor; i'm not sure that it actually duplicates
487  // the stream and copies the data; should that in the BTE
488  // sometimes..
489  data[i] = new AMI_STREAM<T>(*buf.data[i]);
490  deleted[i] = buf.deleted[i];
491  streamsize[i] = buf.streamsize[i];
492 #ifdef SAVE_MEMORY
493  assert(name[i]);
494  delete name[i];
495  name[i] = NULL;
496  name[i] = buf.name[i];
497 #endif
498  }
499  put_streams();
500  MEMORY_LOG("em_buffer: copy constr end\n");
501 }
502 
503 /************************************************************/
504 // free the stream array and the streams pointers
505 template <class T, class Key>
507 {
508 
509  assert(data);
510  // delete the m streams in the buffer
511  get_streams();
512  for (unsigned int i = 0; i < index; i++) {
513  assert(data[i]);
514 #ifdef SAVE_MEMORY
515  check_name(i);
516  delete name[i];
517 #endif
518  delete data[i];
519  data[i] = NULL;
520  }
521 
522  delete[] data;
523  delete[] deleted;
524  delete[] streamsize;
525 #ifdef SAVE_MEMORY
526  delete[] name;
527 #endif
528 }
529 
530 #ifdef SAVE_MEMORY
531 /************************************************************/
532 // checks that name[i] is the same as stream name; stream i must be in
533 // memory (by a previous get_stream call, for instance) in order to
534 // find its length
535 template <class T, class Key>
536 void em_buffer<T, Key>::check_name(unsigned int i UNUSED)
537 {
538 
539 #ifdef EMBUF_CHECK_NAME
540  assert(i >= 0 && i < index);
541  assert(data[i]);
542 
543  char *fooname;
544  data[i]->name(&fooname); // name() allocates the string
545 #ifdef EMBUF_CHECK_NAME_PRINT
546  cout << "::check_name: checking stream [" << level << "," << i
547  << "] name:" << fooname << endl;
548  cout.flush();
549 #endif
550  if (strcmp(name[i], fooname) != 0) {
551  cerr << "name[" << i << "]=" << name[i] << ", streamname=" << fooname
552  << endl;
553  }
554  assert(strcmp(fooname, name[i]) == 0);
555  delete fooname;
556 #endif
557 }
558 #endif
559 
560 /************************************************************/
561 // if SAVE_MEMORY flag is set, load the stream in memory; return the
562 // ith stream
563 template <class T, class Key>
565 {
566 
567  assert(i < index);
568 
569 #ifdef SAVE_MEMORY
570  MY_LOG_DEBUG_ID("em_buffer::get_stream");
571  MY_LOG_DEBUG_ID(i);
572 
573  if (data[i] == NULL) {
574 
575  // stream is on disk, load it in memory
576  assert(name[i]);
577  MY_LOG_DEBUG_ID("load stream in memory");
578  MY_LOG_DEBUG_ID(name[i]);
579 
580 #ifdef EMBUF_PRINT_GETPUT_STREAM
581  cout << "get_stream:: name[" << i << "]=" << name[i] << " from disk\n";
582  cout.flush();
583 #endif
584 
585  // assert that file exists
586  FILE *fp;
587  if ((fp = fopen(name[i], "rb")) == NULL) {
588  cerr << "get_stream: checking that stream " << name[i]
589  << "exists\n";
590  perror(name[i]);
591  assert(0);
592  exit(1);
593  }
594  fclose(fp);
595 
596  // create an AMI_STREAM from file
597  data[i] = new AMI_STREAM<T>(name[i]);
598  assert(data[i]);
599  }
600  else {
601 
602  // if data[i] not NULL, stream must be already in memory
603  MY_LOG_DEBUG_ID("stream not NULL");
604  MY_LOG_DEBUG_ID(data[i]->sprint());
605  }
606 #endif
607 
608  // NOW STREAM IS IN MEMORY
609 
610  // some assertion checks
611  assert(data[i]);
612  assert(data[i]->stream_len() == streamsize[i]);
613 #ifdef SAVE_MEMORY
614  check_name(i);
615 #endif
616 
617  return data[i];
618 }
619 
620 /************************************************************/
621 // if SAVE_MEMORY flag is set, put the i'th stream back on disk
622 template <class T, class Key>
623 void em_buffer<T, Key>::put_stream(unsigned int i)
624 {
625 
626  assert(i < index);
627 
628 #ifdef SAVE_MEMORY
629  MY_LOG_DEBUG_ID("em_buffer::put_stream");
630  MY_LOG_DEBUG_ID(i);
631 
632  if (data[i] != NULL) {
633 
634  // stream is in memory, put it on disk
635  MY_LOG_DEBUG_ID("stream put to disk");
636  MY_LOG_DEBUG_ID(data[i]->sprint());
637 
638  check_name(i);
639 #ifdef EMBUF_PRINT_GETPUT_STREAM
640  cout << "put_stream:: name[" << i << "]=" << name[i] << " to disk\n";
641  cout.flush();
642 #endif
643 
644  // make stream persistent and delete it
645  data[i]->persist(PERSIST_PERSISTENT);
646  delete data[i];
647  data[i] = NULL;
648  }
649  else {
650 
651  // data[i] is NULL, so stream must be already put on disk
652  MY_LOG_DEBUG_ID("stream is NULL");
653  }
654 #endif
655 }
656 
657 /************************************************************/
658 // return a pointer to the streams of the buffer
659 template <class T, class Key>
661 {
662 
663 #ifdef SAVE_MEMORY
664  MY_LOG_DEBUG_ID("em_buffer::get_streams: reading streams from disk");
665 #ifdef EMBUF_PRINT_GETPUT_STREAMS
666  cout << "em_buffer::get_streams (buffer " << level << ")";
667  cout << ": index = " << index << "(arity=" << arity << ")\n";
668  cout.flush();
669 #endif
670 
671  for (unsigned int i = 0; i < index; i++) {
672  get_stream(i);
673  assert(data[i]);
674  }
675 
676 #endif
677 
678  return data;
679 }
680 
681 /************************************************************/
682 // called in pair with load_streams to store streams on disk
683 // and release the memory
684 template <class T, class Key>
686 {
687 
688 #ifdef SAVE_MEMORY
689  MY_LOG_DEBUG_ID("em_buffer::put_streams: writing streams on disk");
690 #ifdef EMBUF_PRINT_GETPUT_STREAMS
691  cout << "em_buffer::put_streams (buffer " << level << ")";
692  cout << ": index = " << index << "(arity=" << arity << ")\n";
693  cout.flush();
694 #endif
695 
696  for (unsigned int i = 0; i < index; i++) {
697  put_stream(i);
698  assert(data[i] == NULL);
699  }
700 #endif
701 }
702 
703 #ifdef SAVE_MEMORY
704 /************************************************************/
705 // return the name of the ith stream
706 template <class T, class Key>
707 char *em_buffer<T, Key>::get_stream_name(unsigned int i) const
708 {
709 
710  assert(i >= 0 && i < index);
711  assert(name[i]);
712  return name[i];
713 }
714 #endif
715 
716 #ifdef SAVE_MEMORY
717 /************************************************************/
718 template <class T, class Key>
720 {
721  unsigned int i;
722  for (i = 0; i < index; i++) {
723  assert(name[i]);
724  cout << "stream " << i << ": " << name[i] << endl;
725  }
726  cout.flush();
727 }
728 #endif
729 
730 /************************************************************/
731 // clean buffer in case some streams have been emptied by deletion
732 template <class T, class Key>
734 {
735 
736  MY_LOG_DEBUG_ID("em_buffer::cleanup()");
737 #ifdef EMBUF_CLEANUP_PRINT
738 #ifdef SAVE_MEMORY
739  if (index > 0) {
740  cout << "before cleanup:\n";
741  print_stream_names();
742  print_stream_sizes();
743  cout.flush();
744  }
745 #endif
746 #endif
747 
748  // load all streams in memory
749  get_streams();
750 
751  // count streams of size=0
752  unsigned int i, empty = 0;
753  for (i = 0; i < index; i++) {
754 
755  if (get_stream_len(i) == 0) {
756  // printing..
757 #ifdef EMBUF_DELETE_STREAM_PRINT
758  cout << "deleting stream [" << level << "," << i << "]:";
759 #ifdef SAVE_MEMORY
760  cout << name[i];
761 #endif
762  cout << endl;
763  cout.flush();
764 #endif
765 
766 #ifdef SAVE_MEMORY
767  // stream is empty ==> delete its name
768  assert(name[i]);
769  delete name[i];
770  name[i] = NULL;
771 #endif
772 
773  // stream is empty ==> reset data
774  assert(data[i]);
775  // data[i]->persist(PERSIST_DELETE); //this is done automatically..
776  delete data[i];
777  data[i] = NULL;
778  deleted[i] = 0;
779  streamsize[i] = 0;
780  empty++;
781  }
782  }
783  // streams are in memory; all streams which are NULL must have been
784  // deleted
785 
786  // shift streams to the left in case holes were introduced
787  unsigned int j = 0;
788  if (empty) {
789 #ifdef EMBUF_DELETE_STREAM_PRINT
790  cout << "em_buffer::cleanup: shifting streams\n";
791  cout.flush();
792 #endif
793  for (i = 0; i < index; i++) {
794  // if i'th stream is not empty, shift it left if necessary
795  if (data[i]) {
796  if (i != j) {
797  // set j'th stream to point to i'th stream
798  // cout << j << " set to " << i << endl; cout.flush();
799  data[j] = data[i];
800  deleted[j] = deleted[i];
801  streamsize[j] = streamsize[i];
802  // set i'th stream to point to NULL
803  data[i] = NULL;
804  deleted[i] = 0;
805  streamsize[i] = 0;
806 #ifdef SAVE_MEMORY
807  // fix the names
808  /* already done assert(name[j]); */
809  /* delete name[j]; */
810  name[j] = name[i];
811  name[i] = NULL;
812  check_name(j);
813 #endif
814  }
815  else {
816  // cout << i << " left the same" << endl;
817  }
818  j++;
819  } // if data[i] != NULL
820  } // for i
821 
822  // set the index
823  assert(index == j + empty);
824  index = j;
825 
826 #ifdef EMBUF_DELETE_STREAM_PRINT
827  cout << "em_buffer::cleanup: index set to " << index << endl;
828  cout.flush();
829 #endif
830  } // if empty
831 
832  // put streams back to disk
833  put_streams();
834 
835 #ifdef EMBUF_CLEANUP_PRINT
836 #ifdef SAVE_MEMORY
837  if (index > 0) {
838  cout << "after cleanup:\n";
839  print_stream_names();
840  print_stream_sizes();
841  cout.flush();
842  }
843 #endif
844 #endif
845 }
846 
847 /************************************************************/
848 // delete all streams
849 template <class T, class Key>
851 {
852 
853  get_streams();
854 
855  // make streams not-persistent and delete them
856  for (unsigned int i = 0; i < index; i++) {
857  assert(data[i]);
858  assert(streamsize[i] == data[i]->stream_len());
859 #ifdef SAVE_MEMORY
860  check_name(i);
861  assert(name[i]);
862  delete name[i];
863  name[i] = NULL;
864 #endif
865 
866  data[i]->persist(PERSIST_DELETE);
867 
868  delete data[i];
869  data[i] = NULL;
870  deleted[i] = 0;
871  streamsize[i] = 0;
872  }
873 
874  index = 0;
875 }
876 
877 /************************************************************/
878 // create and return a stream which contains all elements of
879 // all streams of the buffer in sorted ascending order of
880 // their keys (priorities);
881 template <class T, class Key>
883 {
884 
885  // create stream
886  MEMORY_LOG("em_buffer::sort: allocate new AMI_STREAM\n");
887 
888  AMI_STREAM<T> *sorted_stream =
889  new AMI_STREAM<T>(); /* will be deleted in insert() */
890  assert(sorted_stream);
891 
892  // merge the streams into sorted stream
893  AMI_err aerr;
894  // Key dummykey;
895  // must modify this to seek after deleted[i] elements!!!!!!!!!!!!!
896  // aerr = MIAMI_single_merge_Key(data, arity, sorted_stream,
897  // 0, dummykey);
898  // could not use AMI_merge so i had to write my own..
899 
900  get_streams();
901 
902  aerr = substream_merge(data, arity, sorted_stream);
903  assert(aerr == AMI_ERROR_NO_ERROR);
904 
905  put_streams();
906 
907  return sorted_stream;
908 }
909 
910 /************************************************************/
911 /* merge the input streams; there are <arity> streams in total; write
912  output in <outstream>;
913 
914  the input streams are assumed sorted in increasing order of their
915  keys;
916 
917  assumes the instreams are in memory (no need for get_streams()) */
918 template <class T, class Key>
920  unsigned int arity,
921  AMI_STREAM<T> *outstream)
922 {
923 
924  unsigned int i, j;
925 
926  // some assertion checks
927  assert(instreams);
928  assert(outstream);
929  for (i = 0; i < arity; i++) {
930  assert(instreams[i]);
931 #ifdef SAVE_MEMORY
932  check_name(i);
933 #endif
934  }
935 
936  std::vector<T *> in_objects(
937  arity); // pointers to current leading elements of streams
938  AMI_err ami_err;
939 
940  char str[200];
941  snprintf(str, sizeof(str),
942  "em_buffer::substream_merge: allocate keys array, total %ldB\n",
943  (long)((long)arity * sizeof(merge_key<Key>)));
944  MEMORY_LOG(str);
945 
946  // keys array is initialized with smallest key from each stream (only
947  // non-null keys must be included)
948  merge_key<Key> *keys;
949  // merge_key<Key>* keys = new (merge_key<Key>)[arity];
950  typedef merge_key<Key> footype;
951  keys = new footype[arity];
952  assert(keys);
953 
954  // count number of non-empty streams
955  j = 0;
956  // rewind and read the first item from every stream initializing
957  // in_objects and keys
958  for (i = 0; i < arity; i++) {
959  assert(instreams[i]);
960  // rewind stream
961  if ((ami_err = instreams[i]->seek(deleted[i])) != AMI_ERROR_NO_ERROR) {
962  return ami_err;
963  }
964  // read first item from stream
965  if ((ami_err = instreams[i]->read_item(&(in_objects[i]))) !=
967  if (ami_err == AMI_ERROR_END_OF_STREAM) {
968  in_objects[i] = NULL;
969  }
970  else {
971  return ami_err;
972  }
973  }
974  else {
975  // include this key in the array of keys
976  Key k = in_objects[i]->getPriority();
977  keys[j].set(k, i);
978  j++;
979  }
980  }
981  unsigned int NonEmptyRuns = j;
982 
983  // build heap from the array of keys
984  pqheap_t1<merge_key<Key>> mergeheap(keys, NonEmptyRuns);
985 
986  // repeatedly extract_min from heap, write it to output stream and
987  // insert next element from same stream
988  merge_key<Key> minelt;
989  // rewind output buffer
990  ami_err = outstream->seek(0);
991  assert(ami_err == AMI_ERROR_NO_ERROR);
992  while (!mergeheap.empty()) {
993  // find min key and id of the stream from whereit comes
994  mergeheap.min(minelt);
995  i = minelt.stream_id();
996  // write min item to output stream
997  if ((ami_err = outstream->write_item(*in_objects[i])) !=
999  return ami_err;
1000  }
1001  // read next item from same input stream
1002  if ((ami_err = instreams[i]->read_item(&(in_objects[i]))) !=
1004  if (ami_err != AMI_ERROR_END_OF_STREAM) {
1005  return ami_err;
1006  }
1007  }
1008  // extract the min from the heap and insert next key from same stream
1009  if (ami_err == AMI_ERROR_END_OF_STREAM) {
1010  mergeheap.delete_min();
1011  }
1012  else {
1013  Key k = in_objects[i]->getPriority();
1014  merge_key<Key> nextit(k, i);
1015  mergeheap.delete_min_and_insert(nextit);
1016  }
1017  } // while
1018 
1019  // delete [] keys;
1020  //!!! KEYS BELONGS NOW TO MERGEHEAP, AND WILL BE DELETED BY THE
1021  // DESTRUCTOR OF MERGEHEAP (CALLED AUUTOMATICALLY ON FUNCTION EXIT) IF
1022  // I DELETE KEYS EXPLICITLY, THEY WILL BE DELETED AGAIN BY DESTRUCTOR,
1023  // AND EVERYTHING SCREWS UP..
1024 
1025  return AMI_ERROR_NO_ERROR;
1026 }
1027 
1028 /************************************************************/
1029 // insert an array into the buffer; assume array is sorted; return the
1030 // number of items actually inserted; if SAVE_MEMORY FLAG is on, put
1031 // stream on disk and release its memory
1032 template <class T, class Key>
1033 long em_buffer<T, Key>::insert(T *a, long n)
1034 {
1035 
1036  assert(a);
1037 
1038  if (is_full()) {
1039  cout << "em_buffer::insert: buffer full\n";
1040  return 0;
1041  }
1042 
1043  // can only insert one full stream at a time
1044  // relaxed..
1045  // assert(n == get_stream_maxlen());
1046 
1047  // create the stream
1048  MEMORY_LOG("em_buffer::insert(from array): allocate AMI_STREAM\n");
1049  AMI_STREAM<T> *str = new AMI_STREAM<T>();
1050  assert(str);
1051 
1052  // write the array to stream
1053  AMI_err ae;
1054  for (long i = 0; i < n; i++) {
1055  ae = str->write_item(a[i]);
1056  assert(ae == AMI_ERROR_NO_ERROR);
1057  }
1058  assert(n == str->stream_len());
1059 
1060  // insert the stream in the buffer
1061  return insert(str);
1062 }
1063 
1064 /************************************************************/
1065 /* insert a stream into the buffer; the next free entry in the buffer
1066  is set to point to the stream; if SAVE_MEMORY flag is on, the
1067  stream is put to disk;
1068 
1069  the <nextstream> pointer of buffer is set to point to the argument
1070  stream; (in this way no stream copying is done, just one pointer
1071  copy). The user should be aware the argument stream is 'lost' -
1072  that is a stream which cannot be inserted repeatedly into many
1073  buffers because this would lead to several buffers pointing to the
1074  same stream.
1075 
1076  stream is assume stream is sorted; bos = how many elements must be
1077  skipped (were deleted) from the beginning of stream;
1078 
1079  return the number of items actually inserted */
1080 template <class T, class Key>
1082 {
1083 
1084  assert(str);
1085 
1086  if (is_full()) {
1087  cout << "em_buffer::insert: buffer full\n";
1088  return 0;
1089  }
1090 
1091  // can only insert one level-i-full-stream at a time;
1092  // relaxed..can specify bos;
1093  // not only that, but the length of the stream can be smaller
1094  // than nominal length, because a stream is normally obtained by
1095  // merging streams which can be shorter;
1096  // assert(str->stream_len() == get_stream_len() - bos);
1097 
1098 #ifdef EMBUF_CHECK_INSERT
1099  // check that stream is sorted
1100  cout << "CHECK_INSERT: checking stream is sorted\n";
1101  AMI_err ae;
1102  str->seek(0);
1103  T *crt = NULL, *prev = NULL;
1104  while (str->read_item(&crt)) {
1105  assert(ae == AMI_ERROR_NO_ERROR);
1106  if (prev)
1107  assert(*prev <= *crt);
1108  }
1109 #endif
1110 
1111  // nextstream must be empty
1112  assert(str);
1113  assert(data[nextstream()] == NULL);
1114  assert(deleted[nextstream()] == 0);
1115  assert(streamsize[nextstream()] == 0);
1116 #ifdef SAVE_MEMORY
1117  assert(name[nextstream()] == NULL);
1118 #endif
1119 
1120  // set next entry i the buffer to point to this stream
1121  data[nextstream()] = str;
1122  deleted[nextstream()] = bos;
1123  streamsize[nextstream()] = str->stream_len();
1124 #ifdef SAVE_MEMORY
1125  // set next name entry in buffer to point to this stream's name
1126  char *s;
1127  str->name(&s); // name() allocates the string
1128  name[nextstream()] = s;
1129 
1130  // put stream on disk and release its memory
1132  delete str; // stream should be persistent; just delete it
1133  data[nextstream()] = NULL;
1134 
1135 #ifdef EMBUF_PRINT_INSERT
1136  cout << "insert stream " << s << " at buf [" << level << "," << nextstream()
1137  << "]" << endl;
1138 #endif
1139 #endif
1140 
1141  // increment the index of next available stream in buffer
1142  incr_nextstream();
1143 
1144 #ifdef EMBUF_PRINT_INSERT
1145  print_stream_sizes();
1146  print_stream_names();
1147 #endif
1148 
1149 #ifdef SAVE_MEMORY
1150  MY_LOG_DEBUG_ID("em_buffer::insert(): inserted stream ");
1151  MY_LOG_DEBUG_ID(name[nextstream() - 1]);
1152 #endif
1153 
1154  // return nb of items inserted
1155  return get_stream_len(nextstream() - 1);
1156 }
1157 
1158 /************************************************************/
1159 // print the elements of the i'th stream of the buffer to a stream;
1160 // assumes stream is in memory;
1161 template <class T, class Key>
1162 void em_buffer<T, Key>::print_stream(ostream &s, unsigned int i)
1163 {
1164 
1165  assert(data[i]);
1166  assert((i >= 0) && (i < index));
1167 
1168  AMI_err ae;
1169  T *x;
1170 
1171  s << "STREAM " << i << ": [";
1172 
1173  ae = data[i]->seek(deleted[i]);
1174  assert(ae == AMI_ERROR_NO_ERROR);
1175 
1176  for (long j = 0; j < get_stream_len(i); j++) {
1177  ae = data[i]->read_item(&x);
1178  assert(ae == AMI_ERROR_NO_ERROR);
1179  s << *x << ",";
1180  }
1181  s << "]\n";
1182 }
1183 
1184 /************************************************************/
1185 // print elements range in buffer (read first and last element in each
1186 // substream and find global min and max)
1187 template <class T, class Key>
1189 {
1190 
1191  T *min, *max;
1192  AMI_err ae;
1193 
1194  get_streams();
1195 
1196  for (unsigned int i = 0; i < index; i++) {
1197  cout << "[";
1198  // read min element in substream i
1199  ae = data[i]->seek(deleted[i]);
1200  assert(ae == AMI_ERROR_NO_ERROR);
1201  ae = data[i]->read_item(&min);
1202  assert(ae == AMI_ERROR_NO_ERROR);
1203  cout << min->getPriority() << "..";
1204  // read max element in substream i
1205  ae = data[i]->seek(streamsize[i] - 1);
1206  assert(ae == AMI_ERROR_NO_ERROR);
1207  ae = data[i]->read_item(&max);
1208  assert(ae == AMI_ERROR_NO_ERROR);
1209  cout << max->getPriority() << " (sz=" << get_stream_len(i) << ")] ";
1210  }
1211  for (unsigned int i = index; i < arity; i++) {
1212  cout << "[] ";
1213  }
1214 
1215  put_streams();
1216 }
1217 
1218 /************************************************************/
1219 // print all elements in buffer
1220 template <class T, class Key>
1222 {
1223 
1224  T *x;
1225  AMI_err ae;
1226 
1227  get_streams();
1228 
1229  for (unsigned int i = 0; i < index; i++) {
1230  cout << " [";
1231  ae = data[i]->seek(deleted[i]);
1232  assert(ae == AMI_ERROR_NO_ERROR);
1233  for (unsigned long j = 0; j < get_stream_len(i); j++) {
1234  ae = data[i]->read_item(&x);
1235  assert(ae == AMI_ERROR_NO_ERROR);
1236  cout << x->getPriority() << ",";
1237  }
1238  cout << "]" << endl;
1239  }
1240  for (unsigned int i = index; i < arity; i++) {
1241  cout << "[] ";
1242  }
1243 
1244  put_streams();
1245 }
1246 
1247 /************************************************************/
1248 // print the sizes of the substreams in the buffer
1249 template <class T, class Key>
1251 {
1252 
1253  cout << "(streams=" << index << ") sizes=[";
1254  for (unsigned int i = 0; i < arity; i++) {
1255  cout << get_stream_len(i) << ",";
1256  }
1257  cout << "]" << endl;
1258  cout.flush();
1259 }
1260 
1261 #endif
@ PERSIST_DELETE
Definition: ami_stream.h:114
@ PERSIST_PERSISTENT
Definition: ami_stream.h:116
AMI_err
Definition: ami_stream.h:83
@ AMI_ERROR_END_OF_STREAM
Definition: ami_stream.h:86
@ AMI_ERROR_NO_ERROR
Definition: ami_stream.h:84
#define NULL
Definition: ccmath.h:32
AMI_err write_item(const T &elt)
Definition: ami_stream.h:588
AMI_err name(char **stream_name)
Definition: ami_stream.h:426
AMI_err seek(off_t offset)
Definition: ami_stream.h:445
AMI_err read_item(T **elt)
Definition: ami_stream.h:525
off_t stream_len(void)
Definition: ami_stream.h:375
void persist(persistence p)
Definition: ami_stream.h:639
unsigned int laststream() const
Definition: embuffer.h:254
void cleanup()
Definition: embuffer.h:733
void incr_deleted(unsigned int i)
Definition: embuffer.h:279
long insert(T *a, long n)
Definition: embuffer.h:1033
void print_stream_sizes()
Definition: embuffer.h:1250
unsigned short get_level() const
Definition: embuffer.h:235
unsigned int nextstream() const
Definition: embuffer.h:257
void put_streams()
Definition: embuffer.h:685
void print_range()
Definition: embuffer.h:1188
em_buffer(const unsigned short i, const unsigned long bs, const unsigned int ar)
Definition: embuffer.h:409
unsigned long get_stream_len(unsigned int i)
Definition: embuffer.h:294
~em_buffer()
Definition: embuffer.h:506
long total_deleted() const
Definition: embuffer.h:269
AMI_STREAM< T > * sort()
Definition: embuffer.h:882
void reset()
Definition: embuffer.h:850
unsigned long get_buf_maxlen()
Definition: embuffer.h:312
friend ostream & operator<<(ostream &s, em_buffer &b)
Definition: embuffer.h:364
bool is_empty()
Definition: embuffer.h:315
unsigned long get_buf_len()
Definition: embuffer.h:302
unsigned long get_stream_maxlen() const
Definition: embuffer.h:287
bool is_full() const
Definition: embuffer.h:318
unsigned int get_nbstreams() const
Definition: embuffer.h:263
void incr_nextstream()
Definition: embuffer.h:260
AMI_STREAM< T > ** get_streams()
Definition: embuffer.h:660
long * get_bos() const
Definition: embuffer.h:251
void print()
Definition: embuffer.h:1221
void put_stream(unsigned int i)
Definition: embuffer.h:623
unsigned int get_arity() const
Definition: embuffer.h:266
AMI_STREAM< T > * get_stream(unsigned int i)
Definition: embuffer.h:564
friend int operator!=(const merge_key &x, const merge_key &y)
Definition: embuffer.h:132
KEY key() const
Definition: embuffer.h:108
friend int operator<=(const merge_key &x, const merge_key &y)
Definition: embuffer.h:120
merge_key()
Definition: embuffer.h:97
friend merge_key operator+(const merge_key &x, const merge_key &y UNUSED)
Definition: embuffer.h:140
KEY k
Definition: embuffer.h:93
void set(const KEY &x, const unsigned int sid)
Definition: embuffer.h:103
friend int operator==(const merge_key &x, const merge_key &y)
Definition: embuffer.h:136
~merge_key()
Definition: embuffer.h:101
unsigned int stream_id() const
Definition: embuffer.h:109
friend ostream & operator<<(ostream &s, const merge_key< KEY > &x)
Definition: embuffer.h:112
unsigned int str_id
Definition: embuffer.h:94
KEY getPriority() const
Definition: embuffer.h:110
friend int operator>=(const merge_key &x, const merge_key &y)
Definition: embuffer.h:128
merge_key(const KEY &x, const unsigned int sid)
Definition: embuffer.h:99
friend int operator<(const merge_key &x, const merge_key &y)
Definition: embuffer.h:116
friend int operator>(const merge_key &x, const merge_key &y)
Definition: embuffer.h:124
#define min(x, y)
Definition: draw2.c:29
#define max(x, y)
Definition: draw2.c:30
#define MY_LOG_DEBUG_ID(x)
Definition: embuffer.h:50
#define UNUSED
A macro for an attribute, if attached to a variable, indicating that the variable is not used.
Definition: gis.h:46
#define assert(condition)
Definition: lz4.c:291
void MEMORY_LOG(const std::string &str)
Definition: mm_utils.cpp:59
const char * name
Definition: named_colr.c:6
double b
Definition: r_raster.c:39
#define x