be/src/util/internal-queue-test.cc - impala - Git at Google

 // Licensed to the Apache Software Foundation (ASF) under one
 // or more contributor license agreements.  See the NOTICE file
 // distributed with this work for additional information
 // regarding copyright ownership.  The ASF licenses this file
 // to you 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.


 #include <boost/thread.hpp>
 #include <boost/thread/mutex.hpp>
 #include <glog/logging.h>
 #include <gtest/gtest.h>
 #include <unistd.h>

 #include "common/init.h"
 #include "common/atomic.h"
 #include "util/internal-queue.h"

 #include "common/names.h"

 namespace impala {

 struct IntNode : public InternalQueue<IntNode>::Node {
   IntNode(int value = 0) : value(value) {}
   int value;
 };

 // Basic single threaded operation.
 TEST(InternalQueue, TestBasic) {
   IntNode one(1);
   IntNode two(2);
   IntNode three(3);
   IntNode four(4);

   InternalQueue<IntNode> list;
   ASSERT_TRUE(list.empty());
   ASSERT_EQ(list.size(), 0);
   ASSERT_TRUE(list.Dequeue() == nullptr);
   ASSERT_TRUE(list.Validate());

   list.Enqueue(&one);
   ASSERT_TRUE(!list.empty());
   ASSERT_EQ(list.size(), 1);
   IntNode* i = list.Dequeue();
   ASSERT_TRUE(i != nullptr);
   ASSERT_TRUE(list.empty());
   ASSERT_EQ(list.size(), 0);
   ASSERT_EQ(i->value, 1);
   ASSERT_TRUE(list.Validate());

   list.Enqueue(&one);
   list.Enqueue(&two);
   list.Enqueue(&three);
   list.Enqueue(&four);
   ASSERT_EQ(list.size(), 4);
   ASSERT_TRUE(list.Validate());

   i = list.Dequeue();
   ASSERT_TRUE(i != nullptr);
   ASSERT_EQ(i->value, 1);
   ASSERT_TRUE(list.Validate());

   i = list.Dequeue();
   ASSERT_TRUE(i != nullptr);
   ASSERT_EQ(i->value, 2);
   ASSERT_TRUE(list.Validate());

   i = list.Dequeue();
   ASSERT_TRUE(i != nullptr);
   ASSERT_EQ(i->value, 3);
   ASSERT_TRUE(list.Validate());

   i = list.Dequeue();
   ASSERT_TRUE(i != nullptr);
   ASSERT_EQ(i->value, 4);
   ASSERT_TRUE(list.Validate());

   list.PushFront(&four);
   list.PushFront(&three);
   list.PushFront(&two);
   list.PushFront(&one);

   IntNode* node = list.head();
   int val = 1;
   while (node != nullptr) {
     ASSERT_EQ(node->value, val);
     node = node->Next();
     ++val;
   }

   node = list.tail();
   val = 4;
   while (node != nullptr) {
     ASSERT_EQ(node->value, val);
     node = node->Prev();
     --val;
   }

   for (int i = 0; i < 4; ++i) {
     node = list.PopBack();
     ASSERT_TRUE(node != nullptr);
     ASSERT_EQ(node->value, 4 - i);
     ASSERT_TRUE(list.Validate());
   }
   ASSERT_TRUE(list.PopBack() == nullptr);
   ASSERT_EQ(list.size(), 0);
   ASSERT_TRUE(list.empty());
 }

 // Add all the nodes and then remove every other one.
 TEST(InternalQueue, TestRemove) {
   vector<IntNode> nodes;
   nodes.resize(100);

   InternalQueue<IntNode> queue;

   queue.Enqueue(&nodes[0]);
   queue.Remove(&nodes[1]);
   ASSERT_TRUE(queue.Validate());
   queue.Remove(&nodes[0]);
   ASSERT_TRUE(queue.Validate());
   queue.Remove(&nodes[0]);
   ASSERT_TRUE(queue.Validate());

   for (int i = 0; i < nodes.size(); ++i) {
     nodes[i].value = i;
     queue.Enqueue(&nodes[i]);
   }

   for (int i = 0; i < nodes.size(); i += 2) {
     queue.Remove(&nodes[i]);
     ASSERT_TRUE(queue.Validate());
   }

   ASSERT_EQ(queue.size(), nodes.size() / 2);
   for (int i = 0; i < nodes.size() / 2; ++i) {
     IntNode* node = queue.Dequeue();
     ASSERT_TRUE(node != nullptr);
     ASSERT_EQ(node->value, i * 2 + 1);
   }
 }

 const int VALIDATE_INTERVAL = 10000;

 // CHECK() is not thread safe so return the result in *failed.
 void ProducerThread(InternalQueue<IntNode>* queue, int num_inserts,
     vector<IntNode>* nodes, AtomicInt32* counter, bool* failed) {
   for (int i = 0; i < num_inserts && !*failed; ++i) {
     // Get the next index to queue.
     int32_t value = counter->Add(1) - 1;
     nodes->at(value).value = value;
     queue->Enqueue(&nodes->at(value));
     if (i % VALIDATE_INTERVAL == 0) {
       if (!queue->Validate()) *failed = true;
     }
   }
 }

 void ConsumerThread(InternalQueue<IntNode>* queue, int num_consumes, int delta,
     vector<int>* results, bool* failed) {
   // Dequeued nodes should be strictly increasing.
   int previous_value = -1;
   for (int i = 0; i < num_consumes && !*failed;) {
     IntNode* node = queue->Dequeue();
     if (node == nullptr) continue;
     ++i;
     if (delta > 0) {
       if (node->value != previous_value + delta) *failed = true;
     } else if (delta == 0) {
       if (node->value <= previous_value) *failed = true;
     }
     results->push_back(node->value);
     previous_value = node->value;
     if (i % VALIDATE_INTERVAL == 0) {
       if (!queue->Validate()) *failed = true;
     }
   }
 }

 TEST(InternalQueue, TestClear) {
   vector<IntNode> nodes;
   nodes.resize(100);
   InternalQueue<IntNode> queue;
   queue.Enqueue(&nodes[0]);
   queue.Enqueue(&nodes[1]);
   queue.Enqueue(&nodes[2]);

   queue.Clear();
   ASSERT_TRUE(queue.Validate());
   ASSERT_TRUE(queue.empty());

   queue.PushFront(&nodes[0]);
   queue.PushFront(&nodes[1]);
   queue.PushFront(&nodes[2]);
   ASSERT_TRUE(queue.Validate());
   ASSERT_EQ(queue.size(), 3);
 }

 TEST(InternalQueue, TestSingleProducerSingleConsumer) {
   vector<IntNode> nodes;
   AtomicInt32 counter;
   nodes.resize(1000000);
   vector<int> results;

   InternalQueue<IntNode> queue;
   bool failed = false;
   ProducerThread(&queue, nodes.size(), &nodes, &counter, &failed);
   ConsumerThread(&queue, nodes.size(), 1, &results, &failed);
   ASSERT_TRUE(!failed);
   ASSERT_TRUE(queue.empty());
   ASSERT_EQ(results.size(), nodes.size());

   counter.Store(0);
   results.clear();
   thread producer_thread(ProducerThread, &queue, nodes.size(), &nodes, &counter, &failed);
   thread consumer_thread(ConsumerThread, &queue, nodes.size(), 1, &results, &failed);
   producer_thread.join();
   consumer_thread.join();
   ASSERT_TRUE(!failed);
   ASSERT_TRUE(queue.empty());
   ASSERT_EQ(results.size(), nodes.size());
 }

 TEST(InternalQueue, TestMultiProducerMultiConsumer) {
   vector<IntNode> nodes;
   nodes.resize(1000000);

   bool failed = false;
   for (int num_producers = 1; num_producers < 5; num_producers += 3) {
     AtomicInt32 counter;
     const int NUM_CONSUMERS = 4;
     ASSERT_EQ(nodes.size() % NUM_CONSUMERS, 0);
     ASSERT_EQ(nodes.size() % num_producers, 0);
     const int num_per_consumer = nodes.size() / NUM_CONSUMERS;
     const int num_per_producer = nodes.size() / num_producers;

     vector<vector<int>> results;
     results.resize(NUM_CONSUMERS);

     int expected_delta = -1;
     if (NUM_CONSUMERS == 1 && num_producers == 1) {
       // With one producer and consumer, the queue should have sequential values.
       expected_delta = 1;
     } else if (num_producers == 1) {
       // With one producer, the values added are sequential but can be read off
       // with gaps in each consumer thread.  E.g. thread1 reads: 1, 4, 5, 7, etc.
       // but they should be strictly increasing.
       expected_delta = 0;
     } else {
       // With multiple producers there isn't a guarantee on the order values get
       // enqueued.
       expected_delta = -1;
     }

     InternalQueue<IntNode> queue;
     thread_group consumers;
     thread_group producers;

     for (int i = 0; i < num_producers; ++i) {
       producers.add_thread(
           new thread(ProducerThread, &queue, num_per_producer, &nodes, &counter, &failed));
     }

     for (int i = 0; i < NUM_CONSUMERS; ++i) {
       consumers.add_thread(new thread(ConsumerThread,
           &queue, num_per_consumer, expected_delta, &results[i], &failed));
     }

     producers.join_all();
     consumers.join_all();
     ASSERT_TRUE(queue.empty());
     ASSERT_TRUE(!failed);

     vector<int> all_results;
     for (int i = 0; i < NUM_CONSUMERS; ++i) {
       ASSERT_EQ(results[i].size(), num_per_consumer);
       all_results.insert(all_results.end(), results[i].begin(), results[i].end());
     }
     ASSERT_EQ(all_results.size(), nodes.size());
     sort(all_results.begin(), all_results.end());
     for (int i = 0; i < all_results.size(); ++i) {
       ASSERT_EQ(i, all_results[i]) << all_results[i -1] << " " << all_results[i + 1];
     }
   }
 }

 }

 int main(int argc, char **argv) {
 #ifdef ADDRESS_SANITIZER
   // These tests are disabled for address sanitizer builds.
   // TODO: investigate why the multithreaded ones fail in boost:thread_local_data.
   cerr << "Internal Queue Test Skipped" << endl;
   return 0;
 #endif
   ::testing::InitGoogleTest(&argc, argv);
   impala::InitCommonRuntime(argc, argv, false, impala::TestInfo::BE_TEST);
   return RUN_ALL_TESTS();
 }
	// Licensed to the Apache Software Foundation (ASF) under one
	// or more contributor license agreements. See the NOTICE file
	// distributed with this work for additional information
	// regarding copyright ownership. The ASF licenses this file
	// to you 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.


	#include <boost/thread.hpp>
	#include <boost/thread/mutex.hpp>
	#include <glog/logging.h>
	#include <gtest/gtest.h>
	#include <unistd.h>

	#include "common/init.h"
	#include "common/atomic.h"
	#include "util/internal-queue.h"

	#include "common/names.h"

	namespace impala {

	struct IntNode : public InternalQueue<IntNode>::Node {
	IntNode(int value = 0) : value(value) {}
	int value;
	};

	// Basic single threaded operation.
	TEST(InternalQueue, TestBasic) {
	IntNode one(1);
	IntNode two(2);
	IntNode three(3);
	IntNode four(4);

	InternalQueue<IntNode> list;
	ASSERT_TRUE(list.empty());
	ASSERT_EQ(list.size(), 0);
	ASSERT_TRUE(list.Dequeue() == nullptr);
	ASSERT_TRUE(list.Validate());

	list.Enqueue(&one);
	ASSERT_TRUE(!list.empty());
	ASSERT_EQ(list.size(), 1);
	IntNode* i = list.Dequeue();
	ASSERT_TRUE(i != nullptr);
	ASSERT_TRUE(list.empty());
	ASSERT_EQ(list.size(), 0);
	ASSERT_EQ(i->value, 1);
	ASSERT_TRUE(list.Validate());

	list.Enqueue(&one);
	list.Enqueue(&two);
	list.Enqueue(&three);
	list.Enqueue(&four);
	ASSERT_EQ(list.size(), 4);
	ASSERT_TRUE(list.Validate());

	i = list.Dequeue();
	ASSERT_TRUE(i != nullptr);
	ASSERT_EQ(i->value, 1);
	ASSERT_TRUE(list.Validate());

	i = list.Dequeue();
	ASSERT_TRUE(i != nullptr);
	ASSERT_EQ(i->value, 2);
	ASSERT_TRUE(list.Validate());

	i = list.Dequeue();
	ASSERT_TRUE(i != nullptr);
	ASSERT_EQ(i->value, 3);
	ASSERT_TRUE(list.Validate());

	i = list.Dequeue();
	ASSERT_TRUE(i != nullptr);
	ASSERT_EQ(i->value, 4);
	ASSERT_TRUE(list.Validate());

	list.PushFront(&four);
	list.PushFront(&three);
	list.PushFront(&two);
	list.PushFront(&one);

	IntNode* node = list.head();
	int val = 1;
	while (node != nullptr) {
	ASSERT_EQ(node->value, val);
	node = node->Next();
	++val;
	}

	node = list.tail();
	val = 4;
	while (node != nullptr) {
	ASSERT_EQ(node->value, val);
	node = node->Prev();
	--val;
	}

	for (int i = 0; i < 4; ++i) {
	node = list.PopBack();
	ASSERT_TRUE(node != nullptr);
	ASSERT_EQ(node->value, 4 - i);
	ASSERT_TRUE(list.Validate());
	}
	ASSERT_TRUE(list.PopBack() == nullptr);
	ASSERT_EQ(list.size(), 0);
	ASSERT_TRUE(list.empty());
	}

	// Add all the nodes and then remove every other one.
	TEST(InternalQueue, TestRemove) {
	vector<IntNode> nodes;
	nodes.resize(100);

	InternalQueue<IntNode> queue;

	queue.Enqueue(&nodes[0]);
	queue.Remove(&nodes[1]);
	ASSERT_TRUE(queue.Validate());
	queue.Remove(&nodes[0]);
	ASSERT_TRUE(queue.Validate());
	queue.Remove(&nodes[0]);
	ASSERT_TRUE(queue.Validate());

	for (int i = 0; i < nodes.size(); ++i) {
	nodes[i].value = i;
	queue.Enqueue(&nodes[i]);
	}

	for (int i = 0; i < nodes.size(); i += 2) {
	queue.Remove(&nodes[i]);
	ASSERT_TRUE(queue.Validate());
	}

	ASSERT_EQ(queue.size(), nodes.size() / 2);
	for (int i = 0; i < nodes.size() / 2; ++i) {
	IntNode* node = queue.Dequeue();
	ASSERT_TRUE(node != nullptr);
	ASSERT_EQ(node->value, i * 2 + 1);
	}
	}

	const int VALIDATE_INTERVAL = 10000;

	// CHECK() is not thread safe so return the result in *failed.
	void ProducerThread(InternalQueue<IntNode>* queue, int num_inserts,
	vector<IntNode>* nodes, AtomicInt32* counter, bool* failed) {
	for (int i = 0; i < num_inserts && !*failed; ++i) {
	// Get the next index to queue.
	int32_t value = counter->Add(1) - 1;
	nodes->at(value).value = value;
	queue->Enqueue(&nodes->at(value));
	if (i % VALIDATE_INTERVAL == 0) {
	if (!queue->Validate()) *failed = true;
	}
	}
	}

	void ConsumerThread(InternalQueue<IntNode>* queue, int num_consumes, int delta,
	vector<int>* results, bool* failed) {
	// Dequeued nodes should be strictly increasing.
	int previous_value = -1;
	for (int i = 0; i < num_consumes && !*failed;) {
	IntNode* node = queue->Dequeue();
	if (node == nullptr) continue;
	++i;
	if (delta > 0) {
	if (node->value != previous_value + delta) *failed = true;
	} else if (delta == 0) {
	if (node->value <= previous_value) *failed = true;
	}
	results->push_back(node->value);
	previous_value = node->value;
	if (i % VALIDATE_INTERVAL == 0) {
	if (!queue->Validate()) *failed = true;
	}
	}
	}

	TEST(InternalQueue, TestClear) {
	vector<IntNode> nodes;
	nodes.resize(100);
	InternalQueue<IntNode> queue;
	queue.Enqueue(&nodes[0]);
	queue.Enqueue(&nodes[1]);
	queue.Enqueue(&nodes[2]);

	queue.Clear();
	ASSERT_TRUE(queue.Validate());
	ASSERT_TRUE(queue.empty());

	queue.PushFront(&nodes[0]);
	queue.PushFront(&nodes[1]);
	queue.PushFront(&nodes[2]);
	ASSERT_TRUE(queue.Validate());
	ASSERT_EQ(queue.size(), 3);
	}

	TEST(InternalQueue, TestSingleProducerSingleConsumer) {
	vector<IntNode> nodes;
	AtomicInt32 counter;
	nodes.resize(1000000);
	vector<int> results;

	InternalQueue<IntNode> queue;
	bool failed = false;
	ProducerThread(&queue, nodes.size(), &nodes, &counter, &failed);
	ConsumerThread(&queue, nodes.size(), 1, &results, &failed);
	ASSERT_TRUE(!failed);
	ASSERT_TRUE(queue.empty());
	ASSERT_EQ(results.size(), nodes.size());

	counter.Store(0);
	results.clear();
	thread producer_thread(ProducerThread, &queue, nodes.size(), &nodes, &counter, &failed);
	thread consumer_thread(ConsumerThread, &queue, nodes.size(), 1, &results, &failed);
	producer_thread.join();
	consumer_thread.join();
	ASSERT_TRUE(!failed);
	ASSERT_TRUE(queue.empty());
	ASSERT_EQ(results.size(), nodes.size());
	}

	TEST(InternalQueue, TestMultiProducerMultiConsumer) {
	vector<IntNode> nodes;
	nodes.resize(1000000);

	bool failed = false;
	for (int num_producers = 1; num_producers < 5; num_producers += 3) {
	AtomicInt32 counter;
	const int NUM_CONSUMERS = 4;
	ASSERT_EQ(nodes.size() % NUM_CONSUMERS, 0);
	ASSERT_EQ(nodes.size() % num_producers, 0);
	const int num_per_consumer = nodes.size() / NUM_CONSUMERS;
	const int num_per_producer = nodes.size() / num_producers;

	vector<vector<int>> results;
	results.resize(NUM_CONSUMERS);

	int expected_delta = -1;
	if (NUM_CONSUMERS == 1 && num_producers == 1) {
	// With one producer and consumer, the queue should have sequential values.
	expected_delta = 1;
	} else if (num_producers == 1) {
	// With one producer, the values added are sequential but can be read off
	// with gaps in each consumer thread. E.g. thread1 reads: 1, 4, 5, 7, etc.
	// but they should be strictly increasing.
	expected_delta = 0;
	} else {
	// With multiple producers there isn't a guarantee on the order values get
	// enqueued.
	expected_delta = -1;
	}

	InternalQueue<IntNode> queue;
	thread_group consumers;
	thread_group producers;

	for (int i = 0; i < num_producers; ++i) {
	producers.add_thread(
	new thread(ProducerThread, &queue, num_per_producer, &nodes, &counter, &failed));
	}

	for (int i = 0; i < NUM_CONSUMERS; ++i) {
	consumers.add_thread(new thread(ConsumerThread,
	&queue, num_per_consumer, expected_delta, &results[i], &failed));
	}

	producers.join_all();
	consumers.join_all();
	ASSERT_TRUE(queue.empty());
	ASSERT_TRUE(!failed);

	vector<int> all_results;
	for (int i = 0; i < NUM_CONSUMERS; ++i) {
	ASSERT_EQ(results[i].size(), num_per_consumer);
	all_results.insert(all_results.end(), results[i].begin(), results[i].end());
	}
	ASSERT_EQ(all_results.size(), nodes.size());
	sort(all_results.begin(), all_results.end());
	for (int i = 0; i < all_results.size(); ++i) {
	ASSERT_EQ(i, all_results[i]) << all_results[i -1] << " " << all_results[i + 1];
	}
	}
	}

	}

	int main(int argc, char **argv) {
	#ifdef ADDRESS_SANITIZER
	// These tests are disabled for address sanitizer builds.
	// TODO: investigate why the multithreaded ones fail in boost:thread_local_data.
	cerr << "Internal Queue Test Skipped" << endl;
	return 0;
	#endif
	::testing::InitGoogleTest(&argc, argv);
	impala::InitCommonRuntime(argc, argv, false, impala::TestInfo::BE_TEST);
	return RUN_ALL_TESTS();
	}