Tidy and thread safe builder and frameset

components/streams/src/builder.cpp

@@ -195,7 +195,6 @@ ForeignBuilder::ForeignBuilder(ftl::data::Pool *pool, int id) : BaseBuilder(pool
 	last_frame_ = 0;
 
 	mspf_ = ftl::timer::getInterval();
-	name_ = "NoName";
 }
 
 ForeignBuilder::ForeignBuilder() : BaseBuilder(), head_(0) {
@@ -205,7 +204,6 @@ ForeignBuilder::ForeignBuilder() : BaseBuilder(), head_(0) {
 	last_frame_ = 0;
 
 	mspf_ = ftl::timer::getInterval();
-	name_ = "NoName";
 }
 
 ForeignBuilder::~ForeignBuilder() {
@@ -278,16 +276,14 @@ LockedFrameSet ForeignBuilder::get(int64_t timestamp, size_t ix) {
 		auto fs = _get(timestamp);
 
 		if (ix >= fs->frames.size()) {
-			//throw FTL_Error("Frame index out-of-bounds - " << ix << "(" << fs->frames.size() << ")");
-
-			// FIXME: This is really dangerous
+			// FIXME: Check that no access to frames can occur without lock
+			UNIQUE_LOCK(fs->smtx, flk);
 			while (fs->frames.size() < size_) {
 				fs->frames.push_back(std::move(pool_->allocate(ftl::data::FrameID(fs->frameset(), + fs->frames.size()), fs->timestamp())));
 			}
 		}
 
 		LockedFrameSet lfs(fs.get(), fs->smtx, [this,fs](ftl::data::FrameSet *d) {
-			// TODO: schedule completed
 			if (fs->isComplete()) {
 				if (bufferSize_ == 0 && !fs->test(ftl::data::FSFlag::STALE)) {
 					UNIQUE_LOCK(mutex_, lk);
@@ -300,54 +296,6 @@ LockedFrameSet ForeignBuilder::get(int64_t timestamp, size_t ix) {
 	}
 }
 
-/*void Builder::completed(int64_t ts, size_t ix) {
-	std::shared_ptr<ftl::data::FrameSet> fs;
-
-	{
-		UNIQUE_LOCK(mutex_, lk);
-		fs = _findFrameset(ts);
-	}
-
-	if (fs && ix == 255) {
-		// Note: Frameset can't complete without frames
-		//if (bufferSize_ == 0 && !fs->test(ftl::data::FSFlag::STALE) && static_cast<unsigned int>(fs->count) >= size_) {
-		//	UNIQUE_LOCK(mutex_, lk);
-		//	_schedule();
-		//}
-	} else if (fs && ix < fs->frames.size()) {
-		{
-			UNIQUE_LOCK(fs->mutex(), lk2);
-
-			// If already completed for given frame, then skip
-			if (fs->mask & (1 << ix)) return;
-
-			//states_[ix] = fs->frames[ix].origin();
-			fs->mask |= (1 << ix);
-			++fs->count;
-		}
-
-		//LOG(INFO) << "COMPLETE FRAME : " << fs->timestamp() << " " << fs->count << "(" << size_ << ")";
-
-		// No buffering, so do a schedule here for immediate effect
-		if (bufferSize_ == 0 && !fs->test(ftl::data::FSFlag::STALE) && static_cast<unsigned int>(fs->count) >= size_) {
-			UNIQUE_LOCK(mutex_, lk);
-			_schedule();
-		}
-	} else {
-		LOG(ERROR) << "Completing frame that does not exist: " << ts << ":" << ix;
-	}
-}*/
-
-/*void Builder::markPartial(int64_t ts) {
-	std::shared_ptr<ftl::data::FrameSet> fs;
-
-	{
-		UNIQUE_LOCK(mutex_, lk);
-		fs = _findFrameset(ts);
-		if (fs) fs->set(ftl::data::FSFlag::PARTIAL);
-	}
-}*/
-
 void ForeignBuilder::_schedule() {
 	if (size_ == 0) return;
 	std::shared_ptr<ftl::data::FrameSet> fs;
@@ -366,7 +314,6 @@ void ForeignBuilder::_schedule() {
 			// Calling onFrameset but without all frames so mark as partial
 			if (static_cast<size_t>(fs->count) < fs->frames.size()) fs->set(ftl::data::FSFlag::PARTIAL);
 
-			//for (auto &f : fs->frames) f.store();
 			fs->store();
 
 			//UNIQUE_LOCK(fs->mutex(), lk2);
@@ -391,32 +338,7 @@ void ForeignBuilder::_schedule() {
 
 }
 
-static void mergeFrameset(ftl::data::FrameSet &f1, ftl::data::FrameSet &f2) {
-	// Prepend all frame encodings in f2 into corresponding frame in f1.
-	/*for (size_t i=0; i<f1.frames.size(); ++i) {
-		if (f2.frames.size() <= i) break;
-		f1.frames[i].mergeEncoding(f2.frames[i]);
-	}*/
-}
-
-/*void Builder::_recordStats(float fps, float latency) {
-	UNIQUE_LOCK(msg_mutex__, lk);
-	latency__ += latency;
-	fps__ += fps;
-	++stats_count__;
-}*/
-
 std::pair<float,float> BaseBuilder::getStatistics() {
-	/*UNIQUE_LOCK(msg_mutex__, lk);
-	if (stats_count__ == 0.0f) return {0.0f,0.0f};
-	fps__ /= float(stats_count__);
-	latency__ /= float(stats_count__);
-	float fps = fps__;
-	float latency = latency__;
-	//LOG(INFO) << name_ << ": fps = " << fps_ << ", latency = " << latency_;
-	fps__ = 0.0f;
-	latency__ = 0.0f;
-	stats_count__ = 0;*/
 	return {-1.0f, -1.0f};
 }
 
@@ -438,8 +360,6 @@ std::shared_ptr<ftl::data::FrameSet> ForeignBuilder::_findFrameset(int64_t ts) {
  * Note: Must occur inside a mutex lock.
  */
 std::shared_ptr<ftl::data::FrameSet> ForeignBuilder::_getFrameset() {
-	//LOG(INFO) << "BUF SIZE = " << framesets_.size();
-
 	auto i = framesets_.begin();
 	int N = bufferSize_;
 
@@ -453,17 +373,17 @@ std::shared_ptr<ftl::data::FrameSet> ForeignBuilder::_getFrameset() {
 
 	if (i != framesets_.end()) {
 		auto f = *i;
-		last_frame_ = f->timestamp();
 		
 		// Lock to force completion of on going construction first
 		UNIQUE_LOCK(f->smtx, slk);
+		last_frame_ = f->timestamp();
 		auto j = framesets_.erase(i);
 		f->set(ftl::data::FSFlag::STALE);
 		slk.unlock();
 
 		int count = 0;
-		// Merge all previous frames
-		// TODO: Remove?
+		// Remove all previous framesets
+		// FIXME: Should do this in reverse order.
 		while (j!=framesets_.end()) {
 			++count;
 
@@ -474,18 +394,8 @@ std::shared_ptr<ftl::data::FrameSet> ForeignBuilder::_getFrameset() {
 				UNIQUE_LOCK(f2->smtx, lk);
 				j = framesets_.erase(j);
 			}
-			mergeFrameset(*f,*f2);
-
-			//LOG(INFO) << "EXTRA REMOVED";
-			//_freeFrameset(f2);
 		}
 
-		//if (count > 0) LOG(INFO) << "COUNT = " << count;
-
-		int64_t now = ftl::timer::get_time();
-		//float framerate = 1000.0f / float(now - last_ts_);
-		//_recordStats(framerate, now - f->timestamp());
-		last_ts_ = now;
 		return f;
 	}
 
@@ -498,7 +408,7 @@ std::shared_ptr<ftl::data::FrameSet> ForeignBuilder::_addFrameset(int64_t timest
 		framesets_.clear();
 	}
 
-	auto newf = std::make_shared<FrameSet>(pool_, ftl::data::FrameID(id_,255), timestamp);
+	auto newf = std::make_shared<FrameSet>(pool_, ftl::data::FrameID(id_,255), timestamp, size_);
 	for (size_t i=0; i<size_; ++i) {
 		newf->frames.push_back(std::move(pool_->allocate(ftl::data::FrameID(id_, i), timestamp)));
 	}
@@ -506,9 +416,7 @@ std::shared_ptr<ftl::data::FrameSet> ForeignBuilder::_addFrameset(int64_t timest
 	newf->count = 0;
 	newf->mask = 0;
 	newf->clearFlags();
-	//newf->frames.resize(size_);
-	newf->pose.setIdentity();
-	//newf->clearData();
+	newf->pose.setIdentity();  // Deprecated?
 
 	// Insertion sort by timestamp
 	for (auto i=framesets_.begin(); i!=framesets_.end(); i++) {
@@ -523,7 +431,3 @@ std::shared_ptr<ftl::data::FrameSet> ForeignBuilder::_addFrameset(int64_t timest
 	framesets_.push_back(newf);
 	return newf;
 }
-
-/*void Builder::setName(const std::string &name) {
-	name_ = name;
-}*/

components/structures/include/ftl/data/new_frameset.hpp

@@ -33,7 +33,7 @@ class FrameSet : public ftl::data::Frame {
 
 
 	public:
-	FrameSet(Pool *ppool, FrameID pid, int64_t ts);
+	FrameSet(Pool *ppool, FrameID pid, int64_t ts, size_t psize=1);
 	~FrameSet();
 
 	//int id=0;

components/structures/src/frameset.cpp

 #include <ftl/data/new_frameset.hpp>
 #include <ftl/data/framepool.hpp>
 
-#include <loguru.hpp>
-
 using ftl::data::Frame;
 using ftl::data::FrameSet;
 
-FrameSet::FrameSet(Pool *ppool, FrameID pid, int64_t ts) :
+FrameSet::FrameSet(Pool *ppool, FrameID pid, int64_t ts, size_t psize) :
 	Frame(ppool->allocate(FrameID(pid.frameset(),255), ts)), mask(0) {
 	
 	flush_count = 0; // Reset flush on store...
+	frames.reserve(psize);
 }
 
 FrameSet::~FrameSet() {
@@ -27,7 +26,7 @@ void ftl::data::FrameSet::completed(size_t ix) {
 		mask |= (1 << ix);
 		++count;
 	} else {
-		LOG(ERROR) << "Completing frame that does not exist: " << timestamp() << ":" << ix;
+		throw FTL_Error("Completing frame that does not exist: " << timestamp() << ":" << ix);
 	}
 }
 
@@ -47,25 +46,15 @@ void ftl::data::FrameSet::resize(size_t s) {
 
 void ftl::data::FrameSet::moveTo(ftl::data::FrameSet &fs) {
 	UNIQUE_LOCK(fs.mutex(), lk);
-	//std::unique_lock<std::mutex> lk(fs.mutex());  // FIXME: was a shared mutex
-
-	//if (fs.frames.size() != frames.size()) {
-		// Assume "this" is correct and "fs" is not.
-		//fs.frames.resize(frames.size());
-	//}
-
 	Frame::moveTo(fs);
 	fs.count = static_cast<int>(count);
 	fs.flags_ = (int)flags_;
 	fs.mask = static_cast<unsigned int>(mask);
 	fs.pose = pose;
-
-	/*for (size_t i=0; i<frames.size(); ++i) {
-		frames[i].moveTo(fs.frames[i]);
-	}*/
-
 	fs.frames = std::move(frames);
 
+	count = 0;
+	mask = 0;
 	set(ftl::data::FSFlag::STALE);
 }
 
@@ -95,7 +84,6 @@ void FrameSet::store() {
 
 	{
 		//UNIQUE_LOCK(smtx, lk);
-
 		for (auto &f : frames) if (f.status() == ftl::data::FrameStatus::CREATED) f.store();
 		ftl::data::Frame::store();
 	}

components/structures/src/new_frame.cpp

@@ -261,7 +261,7 @@ void Frame::store() {
 
 	if (!parent_) return;
 
-	//UNIQUE_LOCK(parent_->mutex(), lk);
+	UNIQUE_LOCK(parent_->mutex(), lk);
 
 	for (auto c : changed_) {
 		if (ftl::data::isPersistent(c.first) && hasOwn(c.first)) {