Using Multithreading for Classification

This code snippet shows how to parallelize the classification of new images on the CPU. This code snippet is in C++ 11 and requires a recent compiler.

C++

#include <atomic>
#include <mutex>
#include <queue>
#include <thread>
#include <condition_variable>
#include <memory>

#include <eVision/EasyDeepLearning/EasyClassify/EClassifier.h>

class ImageQueue
{
public:
  ImageQueue() : complete_(false), maxQueueLength_(10) {}

  /// <summary>
  /// Adds an image to the queue.
  /// </summary>
  void addImage(eVision::EBaseROI* img)
  {
    {
      std::unique_lock<std::mutex> locker(mutex_);
      cv_.wait(locker, [&]() { return queue_.size() < maxQueueLength_; });
      queue_.push(img);
    }

    cv_.notify_one();
  }

  /// <summary>
  /// Takes an image from the queue. Returns a null pointer if the queue is complete and doesn't contain anymore images.
  /// </summary>
  eVision::EBaseROI* takeImage()
  {
    eVision::EBaseROI* img = nullptr;
    {
      std::unique_lock<std::mutex> locker(mutex_);
      cv_.wait(locker, [&]() { return (queue_.size() > 0) || isComplete(); });

      if (queue_.size() > 0)
      {
        img = queue_.front();
        queue_.pop();
      }
    }

    cv_.notify_one();
    return img;
  }

  /// <summary>
  /// Indicates that the queue will not received any new images.
  /// </summary>
  void setComplete()
  {
    complete_ = true;
  }

  /// <summary>
  /// Checks whether the queue is empty and will not receive any new images.
  /// </summary>
  bool isComplete() const
  {
    return queue_.size() == 0 && complete_;
  }

private:
  std::atomic_bool complete_;
  std::queue<eVision::EBaseROI*> queue_;
  std::mutex mutex_;
  std::condition_variable cv_;
  int maxQueueLength_;
};

/// <summary>
/// Processing image loop.
/// </summary>
/// <param name="classifierPath">Path to the classifier used to process images</param>
/// <param name="queue">Pointer to the image queue in which to take images.</param>
void processImage(std::string classifierPath, ImageQueue* queue)
{
  using namespace eVision;

  // Instantiate and load the segmenter.
  EasyDeepLearning::EClassifier classifier;
  classifier.Load(classifierPath);

  // Configure the segmenter.
  classifier.SetActiveDeviceByName("CPU");
  classifier.SetBatchSize(32);

  // Loop until there are no more image to process
  while (!queue->isComplete())
  {
    // Take an image from the queue.
    EBaseROI* img = queue->takeImage();

    // If the image is null, continue the loop (and hence, check that the queue is complete).
    if (!img)
      continue;

    EasyDeepLearning::EClassificationResult r = classifier.Classify(*img);

    // Perform operations based on the result.
    // ...

    // Delete the image once it is processed.
    delete img;
  }
}

#define NUM_THREADS 3

using namespace eVision;

// Create an image queue.
std::unique_ptr<ImageQueue> queue(new ImageQueue());

// Declare the vector that will contain our processing threads.
std::vector<std::thread> threads;

// Create the threads
for (int i = 0; i < NUM_THREADS; i++)
{
  threads.emplace_back(processImage, "classifier.ecl", queue.get());
}

// Image acquisition loop
bool hasImage = true;
while (hasImage)
{
  EBaseROI* img = nullptr;

  // Load/Create/Acquire new image in img
  // ...

  queue->addImage(img);
}

// Image acquisition is finished.
queue->setComplete();

// Wait for the thread to finish
for (int i = 0; i < NUM_THREADS; i++)
{
  threads[i].join();
}

C#

using System;
using System.Collections.Concurrent;
using System.Threading;

using eVision;
using eVision.EasyDeepLearning;

static void ClassificationLoop(Object obj)
{
  BlockingCollection<EImageC24> queue = obj as BlockingCollection<EImageC24>;

  EClassifier c = new EClassifier();
  c.ActiveDeviceByName = "CPU";
  c.Load("classifier.ecl");

  while (!queue.IsCompleted)
  {
    EImageC24 image = queue.Take();

    EClassificationResult result = c.Classify(image);
    // Get the most probable label
    string label = result.BestLabel;
    float probability = result.BestProbability;

    // Perform other actions based on the result
  }
}

int NUM_THREADS = 2;

// Queue holding the image to classify
BlockingCollection<EImageC24> imageQueue = new BlockingCollection<EImageC24>(new ConcurrentQueue<EImageC24>(), 2 * NUM_THREADS);

// Create and start the thread pool
Thread[] threads = new Thread[NUM_THREADS];
for (int i = 0; i < NUM_THREADS; i++)
{
  threads[i] = new Thread(ClassificationLoop);
  threads[i].Start(imageQueue);
}

bool hasImage = true;
while (hasImage)
{
  EImageC24 image = new EImageC24();

  // Load or set the data pointer of the image

  // Add the image to the queue
  imageQueue.Add(image);

  // Check that we still have an image to process and change the status
  // of "hasImage" if necessary.
}

// Tell the threads that they won't have any new image coming.
imageQueue.CompleteAdding();

// Wait for the threads to finish
for (int i = 0; i < NUM_THREADS; i++)
  threads[i].Join();

Python

import queue
import threading

import eVision
from eVision import EasyDeepLearning as edl

q = queue.Queue()

def processImage(classifierPath: str):
    classifier = edl.EClassifier()
    classifier.Load(classifierPath)

    classifier.ActiveDeviceByName = "CPU"
    classifier.BatchSize = 32

    while True:
        image = q.get()

        r = classifier.Classify(image)

        # Perform operations based on the result
        # ...

    num_threads = 2
    for i in range(num_threads):
        threading.Thread(target=processImage, daemon=True).start()

    has_image = True
    while has_image:
        image = eVision.EImageC24()

        # Load or set the data point of the image
        # ...

        # Add the image to the queue
        q.put(image)

        # Check that we still have an image to process and change the status
        # of "has_image" if necessary.

    # Block until all tasks are done.
    q.join()