CareerCup

/*
A producer continuously produces a stream of characters. 
There are multiple consumer threads which read the chars
and match to one of known strings and increment the counter
for that pattern. Write the consumer function. 
Show how you will synchronize and maintain parallelism. 

Ex: Producer: abcdegabccaabbaacv ...... 
Known strings[] = {"ab", "aa", "fff" ... } 
patternMatchCount[] = {3, 2, 0 ... }

/*
Solution:
1) Clarification of patternMatchcount is required: do overlaping 
   patterns count or not: e.g. in "AAA" how many times does "AA"
   occure 1 or 2 times? assumption: 2, overlaping counts

2) The syncronization can depend on the pattern matching approach:
   a) The use of brute force or Rabin-Karp requires to hold
      a buffer of length of the pattern to verify if it's a
	  match.
	  Such a buffer can be held by each consumer-thread which
	  can be efficient in terms of synchronization but inefficient
	  in terms of memory usage (if there are a lot of patterns)
   b) If it's automaton based, no buffer is required. Automaton
      based is preferable, but much more complicated to implement
	  (I doubt it is expected to derive the algorithm in an 
	  interview...)

3) For simplicity I would assume the first implementation is that
   each consumer has his own buffer and later on I would correct
   this with an automaton based approach.

4) Synchronization in this case is now, the producer has to 
   push a character to all consumers, this can be done using
   a waitable queue. The waitable queue will be inefficient 
   in terms of syncrhonization needed and because it may 
   dynamically create items, thus a critical section is 
   held over a heap allocation etc...   
   
   It could be improved
   with a lockless buffer which is going to be quite tricky 
   to implement in a non-managed language such as C++ if 
   multicore support is required. If you're not a hard core
   C++ low level programmer with experience in writing drivers
   and lockless stuff it's maybe a good idea to point out 
   that you wont get it error-free and efficient.

5) An other thing to consider is how the system shuts down
   once the producer decides to terminate (reaches the end of 
   the input stream)  

   Here the rather messy code in C++ 11
*/

#include <memory>
#include <iostream>
#include <string>
#include <queue>
#include <thread>
#include <mutex>
#include <condition_variable>

using namespace std;

class WaitableQueue
{
private:
	mutex mutex_;
	condition_variable signal_;
	queue<char> queue_;

public:
	void push(char item) 
	{
		{
			unique_lock<mutex> cs(mutex_);
			queue_.push(item);
		}
		signal_.notify_one();
	}

	char pop() 
	{
		unique_lock<mutex> cs(mutex_);
		signal_.wait(cs, [&](){ return queue_.size() > 0; });
		char result = queue_.front(); 
		queue_.pop();
		return result;
	}	
};

class Consumer
{
private:
	WaitableQueue queue_;
	string pattern_;
	string buffer_;
	int bufferEnd_;
	int counter_;
	thread thread_;

public:
	Consumer(const string& pattern) : counter_(0), 
		bufferEnd_(0), pattern_(pattern)
	{ 
		thread_ = thread([&]{
			while(true) {
				char item = queue_.pop();
				if(item == '\0') break; // end signal
				if(matchNext(item)) counter_++;
			}
		});
	}

	WaitableQueue* getQueue() { return &queue_; }	
	int getCounter() const { return counter_; } 
	string getPattern() const { return pattern_; }
	void join() { thread_.join(); }

private:
	bool matchNext(char next) 
	{
		// this match code is inefficient, improvements possible 
		// by using Rabin-Karp or an automaton based approach
		int m = pattern_.length();
		if(buffer_.length() < bufferEnd_ + 1) buffer_.append(1, next);
		else buffer_[bufferEnd_] = next;
		bufferEnd_ = (bufferEnd_ + 1) % m;
		bool match = buffer_.length() == m; 
		for(int i = 0; i < m && match; ++i) 
			match = buffer_[(bufferEnd_ + i) % m] == pattern_[i]; 
		return match;
	}
};

class Producer
{
private:
	vector<WaitableQueue*> queues_;
	string input_;
	thread thread_;

public:
	Producer(const vector<WaitableQueue*>& queues, const string& input) 
		: queues_(queues), input_(input)
	{
		thread_ = thread([&] {
			// send input
			for(auto ch : input_) { 
				for(auto co : queues) co->push(ch);
			}
			
			// signal end of input
			for(auto co : queues) co->push('\0');		
		});
	}

	void join() { thread_.join(); }	
};


int main()
{
	// create consumers
	vector<Consumer*> consumers({
		new Consumer("ab"),
		new Consumer("aa"),
		new Consumer("ffff")
	});
	
	// get queues from consumers
	vector<WaitableQueue*> queues;
	for(auto& co : consumers) queues.push_back(co->getQueue());

	// create producer
	Producer p(queues, "abcdegabccaabbaacv");
	
	// wait for producer
	p.join();	

	//  wait and print result of consumer
	for(auto& co : consumers) {
		co->join();
		cout << "pattern: " << co->getPattern() 
			<< " count: " 
			<< co->getCounter() << endl;
	}
}

class Item:
	def __init__(self, SeekString="_", SeekIndex=0):
		self.SeekString = SeekString
		self.SeekIndex = SeekIndex

def printSeek(l):
	for item in l:
		print '[',item.SeekString,', ',item.SeekIndex,'] '

def fun():
	KnownStrings = ["aa", "ab", "ff"];
	InputStream = "abcdegabccaabbaacv";
	SeekList = []
	CountDict = {item : 0 for item in KnownStrings}
	for inChar in InputStream:
		for SeekItem in SeekList:
			if SeekItem.SeekString != "_":
				if SeekItem.SeekString[SeekItem.SeekIndex] == inChar:
						# case for one item and that is match
					if SeekItem.SeekString[SeekItem.SeekIndex] == SeekItem.SeekString[-1]:
						CountDict[SeekItem.SeekString] += 1
						# SeekList.remove(SeekItem)
						SeekItem.SeekString = "_"
					else :
						# case for more item but match
						SeekItem.SeekIndex += 1
				else:
					# SeekList.remove(SeekItem)
					SeekItem.SeekString = "_"

		# printSeek(SeekList)

		for KnownString in KnownStrings:
			if KnownString[0] == inChar:
				SeekList.append(Item(KnownString, 1))
		# print ''
	print CountDict
	printSeek(SeekList)


fun()

class Item:
	def __init__(self, SeekString="_", SeekIndex=0):
		self.SeekString = SeekString
		self.SeekIndex = SeekIndex

def printSeek(l):
	for item in l:
		print '[',item.SeekString,', ',item.SeekIndex,'] '

def fun():
	KnownStrings = ["aa", "ab", "ff"];
	InputStream = "abcdegabccaabbaacv";
	SeekList = []
	CountDict = {item : 0 for item in KnownStrings}
	for inChar in InputStream:
		for SeekItem in SeekList:
			if SeekItem.SeekString != "_":
				if SeekItem.SeekString[SeekItem.SeekIndex] == inChar:
						# case for one item and that is match
					if SeekItem.SeekString[SeekItem.SeekIndex] == SeekItem.SeekString[-1]:
						CountDict[SeekItem.SeekString] += 1
						# SeekList.remove(SeekItem)
						SeekItem.SeekString = "_"
					else :
						# case for more item but match
						SeekItem.SeekIndex += 1
				else:
					# SeekList.remove(SeekItem)
					SeekItem.SeekString = "_"

		# printSeek(SeekList)

		for KnownString in KnownStrings:
			if KnownString[0] == inChar:
				SeekList.append(Item(KnownString, 1))
		# print ''
	print CountDict
	printSeek(SeekList)


fun()

class Item:
	def __init__(self, SeekString="_", SeekIndex=0):
		self.SeekString = SeekString
		self.SeekIndex = SeekIndex

def printSeek(l):
	for item in l:
		print '[',item.SeekString,', ',item.SeekIndex,'] '

def fun():
	KnownStrings = ["aa", "ab", "ff"];
	InputStream = "abcdegabccaabbaacv";
	SeekList = []
	CountDict = {item : 0 for item in KnownStrings}
	for inChar in InputStream:
		for SeekItem in SeekList:
			if SeekItem.SeekString != "_":
				if SeekItem.SeekString[SeekItem.SeekIndex] == inChar:
						# case for one item and that is match
					if SeekItem.SeekString[SeekItem.SeekIndex] == SeekItem.SeekString[-1]:
						CountDict[SeekItem.SeekString] += 1
						# SeekList.remove(SeekItem)
						SeekItem.SeekString = "_"
					else :
						# case for more item but match
						SeekItem.SeekIndex += 1
				else:
					# SeekList.remove(SeekItem)
					SeekItem.SeekString = "_"

		# printSeek(SeekList)

		for KnownString in KnownStrings:
			if KnownString[0] == inChar:
				SeekList.append(Item(KnownString, 1))
		# print ''
	print CountDict
	printSeek(SeekList)


fun()

class Item:
	def __init__(self, SeekString="_", SeekIndex=0):
		self.SeekString = SeekString
		self.SeekIndex = SeekIndex

def printSeek(l):
	for item in l:
		print '[',item.SeekString,', ',item.SeekIndex,'] '

def fun():
	KnownStrings = ["aa", "ab", "ff"];
	InputStream = "abcdegabccaabbaacv";
	SeekList = []
	CountDict = {item : 0 for item in KnownStrings}
	for inChar in InputStream:
		for SeekItem in SeekList:
			if SeekItem.SeekString != "_":
				if SeekItem.SeekString[SeekItem.SeekIndex] == inChar:
						# case for one item and that is match
					if SeekItem.SeekString[SeekItem.SeekIndex] == SeekItem.SeekString[-1]:
						CountDict[SeekItem.SeekString] += 1
						# SeekList.remove(SeekItem)
						SeekItem.SeekString = "_"
					else :
						# case for more item but match
						SeekItem.SeekIndex += 1
				else:
					# SeekList.remove(SeekItem)
					SeekItem.SeekString = "_"

		# printSeek(SeekList)

		for KnownString in KnownStrings:
			if KnownString[0] == inChar:
				SeekList.append(Item(KnownString, 1))
		# print ''
	print CountDict
	printSeek(SeekList)


fun()

//example with one consuming thread
class MonitorTest
{
    Random rnd = new Random();
    object key = new object();
    bool m_symbolprocessed = true;
    char m_symbol = 'a';
    ushort m_buffer = 0;
    const ushort m_pattern1 = 0x6566;

    private void consumeSymbol()
    {
        lock (key)
        {
            while (false != m_symbolprocessed)
            {
                Console.WriteLine("waiting producer");
                Monitor.Wait(key);
            }
            Thread.Sleep(rnd.Next() % 200);
            //
            //prosessing
            if (m_buffer == m_pattern1)
            {
                Console.WriteLine("Match found");
                Thread.Sleep(2000);
            }
            //end
            //
            m_symbolprocessed = true;
            Monitor.PulseAll(key);
        }
        consumeSymbol();
    }

    private void produceSymbol()
    {
        lock (key)
        {
            while (true != m_symbolprocessed)
            {
                Console.WriteLine("waiting consumer");
                Monitor.Wait(key);
            }
            Thread.Sleep(rnd.Next() % 200);
            //
            //producing
            m_symbol = (char)(rnd.Next() % 4 + 100);
            m_buffer <<= 8;
            m_buffer |= m_symbol;
            Console.WriteLine(m_symbol);
            //end
            //
            m_symbolprocessed = false;
            Monitor.PulseAll(key);
        }
        produceSymbol();
    }

    public void startAllThreads()
    {
        Thread thAdd = new Thread(new ThreadStart(consumeSymbol));
        Thread thSub = new Thread(new ThreadStart(produceSymbol));

        thAdd.Start();
        thSub.Start();

        thAdd.Join();
        thSub.Join();
    }
}

class Program
{
    static void Main(string[] args)
    {
        MonitorTest mt = new MonitorTest();
        mt.startAllThreads();
    }
}

We can simply use blockingQueue and share it across producer and multiple consumer .

We can simply use a BlockingQueue and share it across producer and consumer

Trick question.
Observe the string to counter mapping can be done by (ZoomBA):

strings = { "ab" : 0 , "aa" : 1 , "fff" : 2 , ... }
counters = [ 3, 2, 0, ...]

def increment_counter( s ){
   idx = strings[ s ]
   #atomic{
      // make atomic operation : increment 
      counters[idx] += 1
   }
}

Note that, there is no need to lock anything beyond the increment, as every index access are independent! Only access on the same index are not.