internal-queue-test.cc

Go to the documentation of this file.

// Copyright 2013 Cloudera Inc.
//
// 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.


#include <boost/thread.hpp>
#include <boost/thread/mutex.hpp>
#include <glog/logging.h>
#include <gtest/gtest.h>
#include <unistd.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() == NULL);
  ASSERT_TRUE(list.Validate());

  list.Enqueue(&one);
  ASSERT_TRUE(!list.empty());
  ASSERT_EQ(list.size(), 1);
  IntNode* i = list.Dequeue();
  ASSERT_TRUE(i != NULL);
  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 != NULL);
  ASSERT_EQ(i->value, 1);
  ASSERT_TRUE(list.Validate());

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

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

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

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

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

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

  for (int i = 0; i < 4; ++i) {
    node = list.PopBack();
    ASSERT_TRUE(node != NULL);
    ASSERT_EQ(node->value, 4 - i);
    ASSERT_TRUE(list.Validate());
  }
  ASSERT_TRUE(list.PopBack() == NULL);
  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 != NULL);
    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, AtomicInt<int32_t>* counter, bool* failed) {
  for (int i = 0; i < num_inserts && !*failed; ++i) {
    // Get the next index to queue.
    AtomicInt<int32_t> value = (*counter)++;
    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 == NULL) 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.Enqueue(&nodes[0]);
  queue.Enqueue(&nodes[1]);
  queue.Enqueue(&nodes[2]);
  ASSERT_TRUE(queue.Validate());
  ASSERT_EQ(queue.size(), 3);
}

TEST(InternalQueue, TestSingleProducerSingleConsumer) {
  vector<IntNode> nodes;
  AtomicInt<int32_t> 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 = 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) {
    AtomicInt<int32_t> 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
  google::InitGoogleLogging(argv[0]);
  ::testing::InitGoogleTest(&argc, argv);
  return RUN_ALL_TESTS();
}