-/* ------------------------------------------------\r
-Copyright 2014 AT&T Intellectual Property\r
- Licensed under the Apache License, Version 2.0 (the "License");\r
- you may not use this file except in compliance with the License.\r
- You may obtain a copy of the License at\r
-\r
- http://www.apache.org/licenses/LICENSE-2.0\r
-\r
- Unless required by applicable law or agreed to in writing, software\r
- distributed under the License is distributed on an "AS IS" BASIS,\r
- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\r
- See the License for the specific language governing permissions and\r
- limitations under the License.\r
- ------------------------------------------- */\r
-\r
-#ifndef MERGE_OPERATOR_H\r
-#define MERGE_OPERATOR_H\r
-\r
-#include "host_tuple.h"\r
-#include "base_operator.h"\r
-#include <list>\r
-using namespace std;\r
-\r
-#include <stdio.h>\r
-\r
-\r
-//int last_tb = 0;\r
-\r
-template <class merge_functor>\r
-class merge_operator : public base_operator {\r
-private :\r
- // input channel on which we want to receive the next tuple\r
- // value -1 indicates that we never received a single tuple\r
- // and we are not concerned on which channel tuple will arrive\r
- int wait_channel;\r
-\r
- // list of tuples from one of the channel waiting to be compared\r
- // against tuple from the other channel\r
- list<host_tuple> merge_queue;\r
- int merge_qsize;\r
-\r
- // comparator object used to compare the tuples\r
- merge_functor func;\r
-\r
- // soft limit on queue size - we consider operator blocked on its input\r
- // whenever we reach this soft limit (not used anymore)\r
- size_t soft_queue_size_limit;\r
-\r
- int dropped_cnt;\r
-\r
- // memory footprint for the merge queue in bytes\r
- unsigned int queue_mem;\r
-\r
- // appends tuple to the end of the merge_queue\r
- // if tuple is stack resident, makes it heap resident\r
- void append_tuple(host_tuple& tup) {\r
- if (!tup.heap_resident) {\r
- char* data = (char*)malloc(tup.tuple_size);\r
- memcpy(data, tup.data, tup.tuple_size);\r
- tup.data = data;\r
- tup.heap_resident = true;\r
- }\r
- merge_queue.push_back(tup);\r
- merge_qsize++;\r
- queue_mem += tup.tuple_size;\r
- }\r
-\r
- void purge_queue(int channel, list<host_tuple>& result){\r
- if (merge_queue.empty())\r
- return;\r
- host_tuple top_tuple = merge_queue.front();\r
- // remove all the tuple smaller than arrived tuple\r
- while(func.compare_stored_with_temp_status(top_tuple, 1-channel) <= 0) {\r
- merge_queue.pop_front();\r
- merge_qsize--;\r
- queue_mem -= top_tuple.tuple_size;\r
- if(merge_qsize<0) abort();\r
- func.update_stored_temp_status(top_tuple,channel);\r
- func.xform_tuple(top_tuple);\r
- top_tuple.channel = output_channel;\r
- result.push_back(top_tuple);\r
-\r
- if (merge_queue.empty())\r
- break;\r
-\r
- top_tuple = merge_queue.front();\r
- func.update_stored_temp_status(top_tuple,channel);\r
- }\r
- }\r
-\r
-\r
-\r
-public:\r
-\r
- merge_operator(int schema_handle1, int schema_handle2, size_t size_limit, const char* name ) : base_operator(name), func(schema_handle1) {\r
- wait_channel = -1;\r
- soft_queue_size_limit = size_limit;\r
- merge_qsize = 0;\r
- dropped_cnt = 0;\r
- queue_mem = 0;\r
- }\r
-\r
-\r
- int accept_tuple(host_tuple& tup, list<host_tuple>& result) {\r
- int last_channel;\r
-\r
- if (tup.channel > 1) {\r
- fprintf(stderr, "Illegal channel number %d for two-way merge\n", tup.channel);\r
- return 0;\r
- }\r
-\r
- /* reject "out of order" tuple - we can receive those after forced flush */\r
- func.get_timestamp(tup);\r
- int res = func.compare_with_temp_status(tup.channel);\r
- bool is_temp_tuple = func.temp_status_received(tup);\r
-\r
-/*\r
-// Ignore temp tuples until we can fix their timestamps.\r
-if(is_temp_tuple){\r
- tup.free_tuple();\r
- return 0;\r
-}*/\r
- if (res < 0){ // out of order tuple\r
-\r
- if (++dropped_cnt % 100000 == 0) {\r
- gslog(LOG_ALERT, "%d tuples dropped by %s merge\n", dropped_cnt,get_name());\r
- }\r
- //if(func.print_warnings())\r
- // fprintf(stderr,"Warning: merge %s receives an out-of-order tuple on channel %d.\n", get_name(), tup.channel);\r
- // free tuple memory\r
- tup.free_tuple();\r
- return 0;\r
- } else {\r
-\r
- if (wait_channel < 0 || wait_channel != tup.channel) {\r
- if (wait_channel < 0)\r
- func.update_temp_status(tup);\r
-\r
- if (!is_temp_tuple){\r
- if(func.compare_with_temp_status(1-tup.channel) <= 0){\r
- if (!tup.heap_resident) {\r
- char* data = (char*)malloc(tup.tuple_size);\r
- memcpy(data, tup.data, tup.tuple_size);\r
- tup.data = data;\r
- tup.heap_resident = true;\r
- }\r
- func.xform_tuple(tup);\r
- tup.channel = output_channel;\r
- result.push_back(tup);\r
- }else{\r
- append_tuple(tup); // put arrived tuple in the queue\r
- }\r
- }\r
-// func.update_temp_status_by_slack(tup, 1-tup.channel);\r
-// purge_queue(tup.channel, result);\r
-\r
- wait_channel = 1-tup.channel;\r
- }else{\r
- // If possible, clear tuples from the other queue.\r
- func.update_temp_status(tup);\r
- purge_queue(1-tup.channel, result);\r
-\r
- if(func.compare_with_temp_status(1-tup.channel) <= 0) { // other tuples in the queue are larger that arrived tuple\r
- if (!is_temp_tuple) {\r
- if (!tup.heap_resident) {\r
- char* data = (char*)malloc(tup.tuple_size);\r
- memcpy(data, tup.data, tup.tuple_size);\r
- tup.data = data;\r
- tup.heap_resident = true;\r
- }\r
- func.xform_tuple(tup);\r
- tup.channel = output_channel;\r
- result.push_back(tup);\r
- }\r
- }\r
- else {\r
- if (!is_temp_tuple)\r
- append_tuple(tup); // put arrived tuple in the queue\r
- wait_channel = 1 - tup.channel; // now we want the tuple from other channel\r
- }\r
- }\r
-\r
- }\r
-\r
- // temp status tuples emited by merge don't serve any other purpose\r
- // other than tracing the flow of tuples\r
- if (is_temp_tuple) {\r
- host_tuple temp_tup;\r
- if (!func.create_temp_status_tuple(temp_tup)) {\r
- temp_tup.channel = output_channel;\r
- result.push_back(temp_tup);\r
- }\r
- // clear memory of heap-resident temporal tuples\r
- tup.free_tuple();\r
- }\r
-\r
- if (!merge_qsize)\r
- wait_channel = -1;\r
-\r
- return 0;\r
- }\r
-\r
- int flush(list<host_tuple>& result) {\r
-\r
- if (merge_queue.empty())\r
- return 0;\r
-\r
- host_tuple top_tuple;\r
- list<host_tuple>::iterator iter;\r
- for (iter = merge_queue.begin(); iter != merge_queue.end(); iter++) {\r
- top_tuple = *iter;\r
- func.update_stored_temp_status(top_tuple,top_tuple.channel);\r
- func.xform_tuple(top_tuple);\r
- top_tuple.channel = output_channel;\r
- result.push_back(top_tuple);\r
- }\r
-\r
- queue_mem = 0;\r
-\r
- return 0;\r
- }\r
-\r
- int set_param_block(int sz, void * value) {\r
- return 0;\r
- }\r
-\r
-\r
- int get_temp_status(host_tuple& result) {\r
- result.channel = output_channel;\r
- return func.create_temp_status_tuple(result);\r
- }\r
-\r
-\r
- int get_blocked_status () {\r
- if (merge_qsize> soft_queue_size_limit)\r
- return wait_channel;\r
- else\r
- return -1;\r
- }\r
-\r
- unsigned int get_mem_footprint() {\r
- return queue_mem;\r
- }\r
-\r
-};\r
-\r
-#endif // MERGE_OPERATOR_H\r
+/* ------------------------------------------------
+Copyright 2014 AT&T Intellectual Property
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+ ------------------------------------------- */
+
+#ifndef MERGE_OPERATOR_H
+#define MERGE_OPERATOR_H
+
+#include "host_tuple.h"
+#include "base_operator.h"
+#include <list>
+using namespace std;
+
+#include <stdio.h>
+
+
+//int last_tb = 0;
+
+template <class merge_functor>
+class merge_operator : public base_operator {
+private :
+ // input channel on which we want to receive the next tuple
+ // value -1 indicates that we never received a single tuple
+ // and we are not concerned on which channel tuple will arrive
+ int wait_channel;
+
+ // list of tuples from one of the channel waiting to be compared
+ // against tuple from the other channel
+ list<host_tuple> merge_queue;
+ int merge_qsize;
+
+ // comparator object used to compare the tuples
+ merge_functor func;
+
+ // soft limit on queue size - we consider operator blocked on its input
+ // whenever we reach this soft limit (not used anymore)
+ size_t soft_queue_size_limit;
+
+ int dropped_cnt;
+
+ // memory footprint for the merge queue in bytes
+ unsigned int queue_mem;
+
+ // appends tuple to the end of the merge_queue
+ // if tuple is stack resident, makes it heap resident
+ void append_tuple(host_tuple& tup) {
+ if (!tup.heap_resident) {
+ char* data = (char*)malloc(tup.tuple_size);
+ memcpy(data, tup.data, tup.tuple_size);
+ tup.data = data;
+ tup.heap_resident = true;
+ }
+ merge_queue.push_back(tup);
+ merge_qsize++;
+ queue_mem += tup.tuple_size;
+ }
+
+ void purge_queue(int channel, list<host_tuple>& result){
+ if (merge_queue.empty())
+ return;
+ host_tuple top_tuple = merge_queue.front();
+ // remove all the tuple smaller than arrived tuple
+ while(func.compare_stored_with_temp_status(top_tuple, 1-channel) <= 0) {
+ merge_queue.pop_front();
+ merge_qsize--;
+ queue_mem -= top_tuple.tuple_size;
+ if(merge_qsize<0) abort();
+ func.update_stored_temp_status(top_tuple,channel);
+ func.xform_tuple(top_tuple);
+ top_tuple.channel = output_channel;
+ result.push_back(top_tuple);
+
+ if (merge_queue.empty())
+ break;
+
+ top_tuple = merge_queue.front();
+ func.update_stored_temp_status(top_tuple,channel);
+ }
+ }
+
+
+
+public:
+
+ merge_operator(int schema_handle1, int schema_handle2, size_t size_limit, const char* name ) : base_operator(name), func(schema_handle1) {
+ wait_channel = -1;
+ soft_queue_size_limit = size_limit;
+ merge_qsize = 0;
+ dropped_cnt = 0;
+ queue_mem = 0;
+ }
+
+
+ int accept_tuple(host_tuple& tup, list<host_tuple>& result) {
+ int last_channel;
+
+ if (tup.channel > 1) {
+ fprintf(stderr, "Illegal channel number %d for two-way merge\n", tup.channel);
+ return 0;
+ }
+
+ /* reject "out of order" tuple - we can receive those after forced flush */
+ func.get_timestamp(tup);
+ int res = func.compare_with_temp_status(tup.channel);
+ bool is_temp_tuple = func.temp_status_received(tup);
+
+/*
+// Ignore temp tuples until we can fix their timestamps.
+if(is_temp_tuple){
+ tup.free_tuple();
+ return 0;
+}*/
+ if (res < 0){ // out of order tuple
+
+ if (++dropped_cnt % 100000 == 0) {
+ gslog(LOG_ALERT, "%d tuples dropped by %s merge\n", dropped_cnt,get_name());
+ }
+ //if(func.print_warnings())
+ // fprintf(stderr,"Warning: merge %s receives an out-of-order tuple on channel %d.\n", get_name(), tup.channel);
+ // free tuple memory
+ tup.free_tuple();
+ return 0;
+ } else {
+
+ if (wait_channel < 0 || wait_channel != tup.channel) {
+ if (wait_channel < 0)
+ func.update_temp_status(tup);
+
+ if (!is_temp_tuple){
+ if(func.compare_with_temp_status(1-tup.channel) <= 0){
+ if (!tup.heap_resident) {
+ char* data = (char*)malloc(tup.tuple_size);
+ memcpy(data, tup.data, tup.tuple_size);
+ tup.data = data;
+ tup.heap_resident = true;
+ }
+ func.xform_tuple(tup);
+ tup.channel = output_channel;
+ result.push_back(tup);
+ }else{
+ append_tuple(tup); // put arrived tuple in the queue
+ }
+ }
+// func.update_temp_status_by_slack(tup, 1-tup.channel);
+// purge_queue(tup.channel, result);
+
+ wait_channel = 1-tup.channel;
+ }else{
+ // If possible, clear tuples from the other queue.
+ func.update_temp_status(tup);
+ purge_queue(1-tup.channel, result);
+
+ if(func.compare_with_temp_status(1-tup.channel) <= 0) { // other tuples in the queue are larger that arrived tuple
+ if (!is_temp_tuple) {
+ if (!tup.heap_resident) {
+ char* data = (char*)malloc(tup.tuple_size);
+ memcpy(data, tup.data, tup.tuple_size);
+ tup.data = data;
+ tup.heap_resident = true;
+ }
+ func.xform_tuple(tup);
+ tup.channel = output_channel;
+ result.push_back(tup);
+ }
+ }
+ else {
+ if (!is_temp_tuple)
+ append_tuple(tup); // put arrived tuple in the queue
+ wait_channel = 1 - tup.channel; // now we want the tuple from other channel
+ }
+ }
+
+ }
+
+ // temp status tuples emited by merge don't serve any other purpose
+ // other than tracing the flow of tuples
+ if (is_temp_tuple) {
+ host_tuple temp_tup;
+ if (!func.create_temp_status_tuple(temp_tup)) {
+ temp_tup.channel = output_channel;
+ result.push_back(temp_tup);
+ }
+ // clear memory of heap-resident temporal tuples
+ tup.free_tuple();
+ }
+
+ if (!merge_qsize)
+ wait_channel = -1;
+
+ return 0;
+ }
+
+ int flush(list<host_tuple>& result) {
+
+ if (merge_queue.empty())
+ return 0;
+
+ host_tuple top_tuple;
+ list<host_tuple>::iterator iter;
+ for (iter = merge_queue.begin(); iter != merge_queue.end(); iter++) {
+ top_tuple = *iter;
+ func.update_stored_temp_status(top_tuple,top_tuple.channel);
+ func.xform_tuple(top_tuple);
+ top_tuple.channel = output_channel;
+ result.push_back(top_tuple);
+ }
+
+ queue_mem = 0;
+
+ return 0;
+ }
+
+ int set_param_block(int sz, void * value) {
+ return 0;
+ }
+
+
+ int get_temp_status(host_tuple& result) {
+ result.channel = output_channel;
+ return func.create_temp_status_tuple(result);
+ }
+
+
+ int get_blocked_status () {
+ if (merge_qsize> soft_queue_size_limit)
+ return wait_channel;
+ else
+ return -1;
+ }
+
+ unsigned int get_mem_footprint() {
+ return queue_mem;
+ }
+
+};
+
+#endif // MERGE_OPERATOR_H
Code Review / com / gs-lite.git / blobdiff