#include <ConjunctiveBroadcaster.h>

Inheritance diagram for knowrob::ConjunctiveBroadcaster:

Collaboration diagram for knowrob::ConjunctiveBroadcaster:

Public Member Functions
	ConjunctiveBroadcaster (bool ignoreInconsistentAnswers=true)

	ConjunctiveBroadcaster (bool ignoreInconsistentAnswers=true)

Public Member Functions inherited from knowrob::TokenBroadcaster
	TokenBroadcaster ()

	~TokenBroadcaster () override

void	addSubscriber (const std::shared_ptr< Channel > &subscriber)

void	removeSubscriber (const std::shared_ptr< Channel > &subscriber)

	TokenBroadcaster ()

	~TokenBroadcaster () override

void	addSubscriber (const std::shared_ptr< Channel > &subscriber)

void	removeSubscriber (const std::shared_ptr< Channel > &subscriber)

Public Member Functions inherited from knowrob::TokenStream
	TokenStream ()

virtual	~TokenStream ()

	TokenStream (const TokenStream &)=delete

virtual void	close ()

bool	isOpened () const

	TokenStream ()

virtual	~TokenStream ()

	TokenStream (const TokenStream &)=delete

virtual void	close ()

bool	isOpened () const

Protected Types
using	AnswerMap = std::map< uint32_t, std::map< size_t, AnswerPtr > >

using	AnswerMap = std::map< uint32_t, std::map< size_t, AnswerPtr > >

Protected Member Functions
void	push (Channel &channel, const TokenPtr &tok) override

void	genCombinations (uint32_t pushedChannelID, AnswerMap::iterator it, AnswerPtr &combinedResult)

void	push (Channel &channel, const TokenPtr &tok) override

void	genCombinations (uint32_t pushedChannelID, AnswerMap::iterator it, AnswerPtr &combinedResult)

Protected Member Functions inherited from knowrob::TokenBroadcaster
void	push (const TokenPtr &tok) override

virtual void	pushToBroadcast (const TokenPtr &tok)

void	push (const TokenPtr &tok) override

virtual void	pushToBroadcast (const TokenPtr &tok)

Protected Attributes
AnswerMap	buffer_

std::mutex	buffer_mutex_

bool	ignoreInconsistentAnswers_

bool	hasSolution_

unsigned int	numClosedChannels_ = 0

std::vector< AnswerNoPtr >	negativeAnswers_

Protected Attributes inherited from knowrob::TokenBroadcaster
std::list< std::shared_ptr< Channel > >	subscribers_

std::mutex	mtx_

Protected Attributes inherited from knowrob::TokenStream
std::list< std::shared_ptr< Channel > >	channels_

std::atomic< bool >	isOpened_

std::mutex	channel_mutex_

Detailed Description

Consolidates answers from multiple sources that are considered in conjunction. This is intended to be used for parallel evaluation of independent sub-goals within a query.

Definition at line 19 of file ConjunctiveBroadcaster.h.

Member Typedef Documentation

◆ AnswerMap [1/2]

using knowrob::ConjunctiveBroadcaster::AnswerMap = std::map<uint32_t, std::map<size_t, AnswerPtr> >

protected

Definition at line 26 of file ConjunctiveBroadcaster.h.

◆ AnswerMap [2/2]

using knowrob::ConjunctiveBroadcaster::AnswerMap = std::map<uint32_t, std::map<size_t, AnswerPtr> >

protected

Definition at line 26 of file ConjunctiveBroadcaster.h.

Constructor & Destructor Documentation

◆ ConjunctiveBroadcaster() [1/2]

ConjunctiveBroadcaster::ConjunctiveBroadcaster ( bool ignoreInconsistentAnswers = true )

explicit

Definition at line 15 of file ConjunctiveBroadcaster.cpp.

         : TokenBroadcaster(),
           ignoreInconsistentAnswers_(ignoreInconsistentAnswers),
           hasSolution_(false) {}

◆ ConjunctiveBroadcaster() [2/2]

knowrob::ConjunctiveBroadcaster::ConjunctiveBroadcaster ( bool ignoreInconsistentAnswers = true )

explicit

Member Function Documentation

◆ genCombinations() [1/2]

void ConjunctiveBroadcaster::genCombinations	(	uint32_t	pushedChannelID,
		AnswerMap::iterator	it,
		AnswerPtr &	combinedResult
	)

protected

Definition at line 71 of file ConjunctiveBroadcaster.cpp.

                                                                                    {
     if (it == buffer_.end()) {
         // end reached, push combination
         TokenBroadcaster::push(combinedResult);
         hasSolution_ = true;
     } else if (it->first == pushedChannelID) {
         // pass through channel from which the new message was pushed
         auto it1 = it;
         ++it1;
         genCombinations(pushedChannelID, it1, combinedResult);
     } else if (it->second.size() == 1) {
         // only a single message buffered from this channel
         auto it1 = it;
         ++it1;
         auto merged = mergeAnswers(combinedResult,
                                    it->second.begin()->second, ignoreInconsistentAnswers_);
         if (merged) {
             genCombinations(pushedChannelID, it1, merged);
         }
     } else {
         // generate a combination for each buffered message
         // note: the number of possible combinations grows exponentially with number of messages in channels
         auto it1 = it;
         ++it1;
         for (auto &msg: it->second) {
             auto merged = mergeAnswers(
                     combinedResult, msg.second, ignoreInconsistentAnswers_);
             if (merged) {
                 genCombinations(pushedChannelID, it1, merged);
             }
         }
     }
 }

◆ genCombinations() [2/2]

void knowrob::ConjunctiveBroadcaster::genCombinations	(	uint32_t	pushedChannelID,
		AnswerMap::iterator	it,
		AnswerPtr &	combinedResult
	)

protected

◆ push() [1/2]

void ConjunctiveBroadcaster::push	(	Channel &	channel,
		const TokenPtr &	tok
	)

overrideprotectedvirtual

Reimplemented from knowrob::TokenStream.

Definition at line 20 of file ConjunctiveBroadcaster.cpp.

                                                                        {
     if (tok->tokenType() == TokenType::ANSWER_TOKEN) {
         auto answer = std::static_pointer_cast<const Answer>(tok);
         if (answer->isPositive()) {
             const uint32_t channelID = channel.id();
             // need to lock the whole push as genCombinations uses an iterator over the buffer.
             std::lock_guard<std::mutex> lock(buffer_mutex_);
  
             // add to the buffer for later combinations
             // replace other answer with same hash if present
             buffer_[channelID][answer->hashOfAnswer()] = answer;
  
             // generate combinations with other channels if each channel
             // buffer has some content.
             if (buffer_.size() == channels_.size()) {
                 if (channels_.size() == 1) {
                     // not needed to generate combinations
                     TokenBroadcaster::push(tok);
                 } else {
                     // generate all combinations and push combined messages
                     genCombinations(channelID, buffer_.begin(), answer);
                 }
             }
         } else if (answer->isNegative()) {
             // do not combine negative answers like the positive ones above.
             // only push "no" when receiving EOF while no positive answer has been produced.
             negativeAnswers_.emplace_back(std::static_pointer_cast<const AnswerNo>(answer));
         }
     } else {
         if (tok->indicatesEndOfEvaluation()) {
             numClosedChannels_++;
             if (numClosedChannels_ == channels_.size() && !hasSolution_) {
                 if (negativeAnswers_.size() == 1) {
                     TokenBroadcaster::push(negativeAnswers_.front());
                 } else {
                     auto no = std::make_shared<AnswerNo>();
                     for (auto &x: negativeAnswers_) {
                         no->mergeWith(*x);
                     }
                     TokenBroadcaster::push(no);
                 }
             } else {
                 // do not pass on the EOF token until all channels have finished
                 return;
             }
         }
         // pass through non-answer messages
         TokenStream::push(channel, tok);
     }
 }

◆ push() [2/2]

void knowrob::ConjunctiveBroadcaster::push	(	Channel &	channel,
		const TokenPtr &	tok
	)

overrideprotectedvirtual

Reimplemented from knowrob::TokenStream.

Member Data Documentation

◆ buffer_

AnswerMap knowrob::ConjunctiveBroadcaster::buffer_

protected

Definition at line 28 of file ConjunctiveBroadcaster.h.

◆ buffer_mutex_

std::mutex knowrob::ConjunctiveBroadcaster::buffer_mutex_

protected

Definition at line 29 of file ConjunctiveBroadcaster.h.

◆ hasSolution_

bool knowrob::ConjunctiveBroadcaster::hasSolution_

protected

Definition at line 31 of file ConjunctiveBroadcaster.h.

◆ ignoreInconsistentAnswers_

bool knowrob::ConjunctiveBroadcaster::ignoreInconsistentAnswers_

protected

Definition at line 30 of file ConjunctiveBroadcaster.h.

◆ negativeAnswers_

std::vector< AnswerNoPtr > knowrob::ConjunctiveBroadcaster::negativeAnswers_

protected

Definition at line 33 of file ConjunctiveBroadcaster.h.

◆ numClosedChannels_

unsigned int knowrob::ConjunctiveBroadcaster::numClosedChannels_ = 0

protected

Definition at line 32 of file ConjunctiveBroadcaster.h.

The documentation for this class was generated from the following files:

build/_CPack_Packages/Linux/DEB/knowrob-2.1.0-Linux/usr/include/knowrob/queries/ConjunctiveBroadcaster.h
src/queries/ConjunctiveBroadcaster.cpp

Public Member Functions

Protected Types

Protected Member Functions

Protected Attributes

Detailed Description

Member Typedef Documentation

◆ AnswerMap [1/2]

◆ AnswerMap [2/2]

Constructor & Destructor Documentation

◆ ConjunctiveBroadcaster() [1/2]

◆ ConjunctiveBroadcaster() [2/2]

Member Function Documentation

◆ genCombinations() [1/2]

◆ genCombinations() [2/2]

◆ push() [1/2]

◆ push() [2/2]

Member Data Documentation

◆ buffer_

◆ buffer_mutex_

◆ hasSolution_

◆ ignoreInconsistentAnswers_

◆ negativeAnswers_

◆ numClosedChannels_