[Herz 6]

Тему разделил. Переносить в "Операционные системы" пока не стал. Какие будут мнения?

[Slash 0]

brag, можете показать реализацию variable element-size queue?

 

Как у меня сделано:

очередь выдрана из mbed

#include <stdint.h>
#include <string.h>
#include "cmsis_os.h"
#include "mbed_error.h"

bool handlerMode()
{
return __get_IPSR() != 0;
}

// Очередь с ленивой инициализацией - инициализируется в момент первого использования.
// Сделано для обхода проблемы с неопределенным порядком инициализации глобальных объектов.
// Очереди создаются глобально и менеджер памяти создается глобально.
// Очередь использует выделение памяти в куче, поэтому должна быть создана после менеджера памяти.
// Поэтому и сделана ленивая инициализация.
// Бывают случаи, когда первое использование очереди происходит в прерывании, что недопустимо.
// На этот случай, нужно вызывать метод  create() явно, до первого использования.	
template<typename T, size_t N>
class Queue
{
public:
Queue(const char * name = NULL)
	: mName(name)
{}

/** Put a message in a Queue.
  @param   data	  data.
  @param   millisec  timeout value or 0 in case of no time-out. (default: 0)
  @return  status code that indicates the execution status of the function.
*/
bool put(const T & data, uint32_t millisec = 0)
{
	if (!mQueueId)
	{
		create();
	}
	portBASE_TYPE taskWoken = pdFALSE; // не понятно, зачем это надо
	if (handlerMode())
	{
		if (xQueueSendToBackFromISR(mQueueId, reinterpret_cast<const void *>(&data), &taskWoken) == pdTRUE)
		{
			portEND_SWITCHING_ISR(taskWoken);
			return true;
		}
	}
	else
	{
		TickType_t ticks = millisec / portTICK_PERIOD_MS;

		if (ticks == 0)
		{
			ticks = 1;
		}

		if (xQueueSendToBack(mQueueId, reinterpret_cast<const void *>(&data), ticks) == pdTRUE)
		{
			return true;
		}
	}
	return false;
}

/** Get a message or Wait for a message from a Queue.
  @param   millisec  timeout value or 0 in case of no time-out. (default: osWaitForever).
  @return  event information that includes the message and the status code.
*/
bool hasMessage(uint32_t millisec = osWaitForever)
{
	if (!mQueueId)
	{
		create();
	}
	bool hasMessage = false;
	portBASE_TYPE taskWoken = pdFALSE;
	if (handlerMode())
	{
		hasMessage = xQueueReceiveFromISR(mQueueId, reinterpret_cast<void *>(&mData), &taskWoken) == pdTRUE;
		portEND_SWITCHING_ISR(taskWoken);
	}
	else
	{
		TickType_t ticks = 0;
		if (millisec == osWaitForever)
		{
			ticks = portMAX_DELAY;
		}
		else if (millisec != 0)
		{
			ticks = millisec / portTICK_PERIOD_MS;
			if (ticks == 0)
			{
				ticks = 1;
			}
		}

		hasMessage = xQueueReceive(mQueueId, reinterpret_cast<void *>(&mData), ticks) == pdTRUE;
	}
	return hasMessage;
}

const T & message() const
{
	return mData;
}

T & message()
{
	return mData;
}

size_t spaceAvailable() const
{
	return uxQueueSpacesAvailable(mQueueId);
}

// нельзя вызывать из прерывания.
void create()
{
	// неплохо бы проверять, что мы находимся не в прерывании
	mQueueId = xQueueCreate(N, sizeof(T));

	if (!mQueueId)
	{
		error("Error initialising the queue object\n");
	}
	if (mName)
	{
		vQueueAddToRegistry(mQueueId, mName);
	}
}

private:
const char * mName;
QueueHandle_t mQueueId;
T mData;
};

 

В очередь пихаю свой доморощенный Variant:

#include <algorithm>

template <size_t N>
class Variant
{
public:
Variant()
{}

template <typename T> 
Variant(const T & t)
{
	std::copy(reinterpret_cast<const uint8_t *>(&t), reinterpret_cast<const uint8_t *>(&t) + sizeof(mData), mData);
}

template <typename T>
T get() const
{
	return *reinterpret_cast<const T *>(mData);
}

template <typename T>
T * value() 
{
	return reinterpret_cast<T *>(mData);
}

Variant & operator= (const Variant & other)
{
	std::copy(other.mData, other.mData + sizeof(mData), mData);
	return *this;
}

float toFloat() const
{
	return get<float>();
}

uint32_t toUInt32() const
{
	return get<uint32_t>();
}


private:
uint8_t mData[N];
};

 

Использование:

 

struct Message
{
...
};

typedef Variant<sizeof(Message)> MessageVariant;  // знаю, что структура Message самая большая из помещаемых в MessageVariant
Queue<MessageVariant, 10> queue;

...
if (queue.hasMessage())
{
   const MessageVariant & msgVar = queue.message();
   const Message & msg = msgVar .data().get<Message>();
}

Реализация на моем уровне знания шаблонов.

Изменено 10 сентября, 2016 пользователем Slash

[AHTOXA 18]

Тема про SST интересная. Не панацея, конечно, но попробовать можно. Жаль, что brag не привёл какого-нибудь полного примера своей реализации. Думаю, что с использованием c++11/c++14 должно получиться весьма изящно.

 

На всякий случай приведу ссылку на репозиторий с исходной реализацией SST и его портом под какой-то арм.

[brag 0]

brag, можете показать реализацию variable element-size queue?

Ваша реализация не понравилась, слишком много левого зависимого кода. И куча приведений типов, тем более reinterpret_cast.

Вот моя, без зависимостей вообще. Это самый простой вариант. Текущую версию приводить не буду, чтобы не путались, там локфри, записаь происходит за несколько этапов и сложно для понимания. Разберемся с обычной - перейдем к лок-фри.

Шаблон уже видели, вот его реализация

public:
typedef uint32_t BufferT;

// Fifo item base class. Must be POD
class ItemBase{
public:
	// pointer to function that knows how to call specific lambda
	typedef void(*WrapperT)(ItemBase*, TArgs...);

	void invoke(TArgs ... args){
		wrapper_(this, args...);
	}

	// ------------------------------------------
	static size_t mark_unused(size_t size){
		return size | UnusedMask;
	}

	size_t size()const{ return size_ & ~UnusedMask; }
	bool isUsed()const{ return (size_ & UnusedMask) == 0; }

	enum{ UnusedMask = 1<<(sizeof(size_t)*8-1) };

	// ------------------------------------------
protected:
	size_t size_; // in sizeof(BufferT) units
	WrapperT wrapper_;
};

private:
// Constructors are private.
// You cant create objects, create VdelegateFIFO_T instead
// But You can use pointers of this type
VdelegateFIFO();
VdelegateFIFO(uint16_t size): size(size){
	// empty: rx == wx && pushed_objects == 0
	// full: rx == wx && pushed_objects > 0
	rx = wx = 0;
	pushed_objects = 0;
}


// lambda, function or functor object
template<class T>
class Item : public ItemBase{
public:
	Item(const T& la): lambda(la){
		this->size_ = obect_size();
		this->wrapper_ = [](ItemBase* that, TArgs ... args){
			static_cast<Item*>(that)->lambda(args...);
		};
	}

	// in sizeof(ItemBase) units
	static constexpr size_t obect_size(){
		return sizeof(Item)/sizeof(BufferT);
	}

	void* operator new(size_t, void *ptr){ return ptr; }

private:
	T lambda;
};

// -------------------- public --------------------
public:
template<class T>
bool push(const T& obj){
	int w = getFreeConsecutiveSpace(Item<T>::obect_size());
	if(w<0){ // not enough space
		return false;
	}
	new(&buffer[w]) Item<T>(obj);
	wx = w + Item<T>::obect_size();
	if(wx >= size)wx = 0; // wrap

	pushed_objects++;
	return true;
}

ItemBase* getItemToPop(){
	if(isEmpty())return nullptr;
	ItemBase *item = reinterpret_cast<ItemBase*>(&buffer[rx]);
	// wrap if unused tail
	if(!item->isUsed()){
		rx = 0;
		item = reinterpret_cast<ItemBase*>(&buffer[rx]);
	}
	return item;
}

bool pop(const ItemBase *item){
	if(!item)return false;
	if(isEmpty())return false;
	pushed_objects--;
	rx += item->size();
	if(rx>=size){
		rx=0; // wrap
	}else if(rx>wx){
		// pop unused tail of buffer too
		ItemBase *item = reinterpret_cast<ItemBase*>(&buffer[rx]);
		if(!item->isUsed()){
			//rx += item->size();
			//if(rx >= size)rx=0;
			rx=0; // wrap
		}
	}
	return true;
}

bool isEmpty()const{ return pushed_objects==0; }

// ---------------------------------------------------
private:
int getFreeConsecutiveSpace(int sz){
	int w = wx;
	if(w==rx && pushed_objects>0)return -1; // full
	// wx after rx
	if(w >= rx){
		if(w+sz <= size)return w;
		// mark tail of buffer unused
		buffer[w] = ItemBase::mark_unused(size-w);
		// wrap
		w = 0;
	}
	// wx before rx
	if(w+sz <= rx)return w;
	return -1;
}

template<class T, int Size> friend class VdelegateFIFO_T;

private:
const uint16_t size; // in sizeof(BufferT) units
uint16_t rx, wx;
uint16_t pushed_objects;
// it's a bit tricky and non-standard but works
BufferT buffer[0];
};

// ------------------ Container ------------------
template<class T, int Size_w>
class VdelegateFIFO_T : public VdelegateFIFO<T>{
public:
VdelegateFIFO_T(): VdelegateFIFO<T>(Size_w) {}

private:
typedef typename VdelegateFIFO<T>::BufferT BufferT;
BufferT buffer[size_w];
};

Не знаю, понятен ли этот код, поэтому обясню на пальцах, а там можно и самому сделать.

Выделяется буфер фиксированного размера - это и есть сторейж нашей очереди. Данные пишуться по кругу, ring-buffer. Но с одной оговоркой - если обьект не влазит в конец буфера, этот конец помечается, как unused, а обьект пытаемся поместить в начало буфера. Если и там места нет - значит в очереди нет места для обьекта. Все. Если очередь используется с разных уровней приоритета - нужен лок на всю операцию записи/чтения.

В лок-фри в принципе идея та же, только запись служебных полей и самих данных разделено во времени(на 3 этапа). Еще есть одна реализация, где лок есть только на служебные поля, но запись данных идет без лока, тоже 3-этапная запись. Используется на процах, где нет нужной поддержки(у кортекс это ldrex/strex)

 

Вот пример использования очереди, ne lock-free , обычной:

template<int Size_w> //, class TBase=void*>
class TaskQueue{
public:
TaskQueue(SST::TPriority priority):
	priority(priority)
{
	current_task = nullptr; // dont pop first time
}

template<class T>
bool enqueueTask(const T& la){
	CriticalSection cs;
	cs.enter();
	bool e = fifo.isEmpty();
	bool r = fifo.push(la);
	cs.leave();
	if(e)taskCompleted();
	return r;
}

bool taskCompleted(){
	bool r = SST::postMessage(priority, [this](){
		processNextTask();
	});
	if(!r){
		printf("TaskQueue::taskCompleted error: SST_queue %d full\n", priority);
	}
	return r;
}

private:
void processNextTask(){
	CriticalSection cs;
	cs.enter();
	fifo.pop(current_task); // pop if exists
	auto la = fifo.getItemToPop();
	current_task = la;
	cs.leave();
	if(la)la->invoke(); // static_cast<TBase*>(this));
}

private:
typedef VdelegateFIFO_T<void(), Size_w> T_Fifo;
T_Fifo fifo;
typename T_Fifo::ItemBase* current_task;
const SST::TPriority priority;
};

Мжно использовать из любого уровня приоритта, в том числе из прерываний.

Как пользоваться этим я уже показывал ранее(пример стирания флешки, шина SPI)

 

Тема про SST интересная. Не панацея, конечно, но попробовать можно. Жаль, что brag не привёл какого-нибудь полного примера своей реализации.

Не панацея, но это другой стиль и он имеет много преимуществ перед синхронным, я их всех уже разжевал в этой теме и показал на примерах. Примеров кода в качестве контр-аргументов не увидел. Увижу - попытаюсь сделать из них SST-вариант, посмотрим на какой платформе будет круче и проще :)

Полный пример приводить смысла нет - все равно не разберетесь без детального изучения и практики асинхронного программирования, да и так примеров я тут кучу привел, реально работающих, во всех разобрались и поняли как работает?. Мое ядро ССТ ничем не отличается от того, что по ссылке, только сделано на плюсах и заточено под мои прихоти. Но если что-то конкретное интересует - спрашивайте, покажу.

 

Готового решения у меня нет, если хотите работать на SST - придется реализовывать самому, там очень мало строк кода. Если не сможете реализовать, значит не сможете работать и на готовой реализации.

Если все равно есть желание - практикуйтесь на NodeJS, там все уже готово, и есть отличная документация и куча инфы в интернете. Но там 1 приоритет, но это даже хорошо - научитесь работать в одном приоритете - потом сможете освоить и много-приоритетный стиль и движок свой напишете за пару дней.

[AHTOXA 18]

Готового решения у меня нет, если хотите работать на SST - придется реализовывать самому, там очень мало строк кода.

Как-то не очень правильно для миссионера предлагать для "попробовать" реализовать всё самому:)

Первая доза должна быть бесплатно! :))

А если серьёзно - когда есть возможность погонять готовую реализацию, понимание приходит гораздо быстрее. К тому же лично мне гораздо проще разобраться в целом работающем проекте, чем в статье с кусками кода.

[brag 0]

Да я и не миссионер и готовой реализации у меня нет, у меня есть свой никому не нужный велосипед, применить кроме меня кто-то другой вряд ли сможет, а если сможет, то ему будет проще сделать свой. А готовая реализация - это NodeJS, с него и надо начинать.

Целый работающий проект - коммерческий и права на него не принадлежат мне, я не имею права его публиковать.

[AHTOXA 18]

Вы тут в пылу борьбы уже столько проектов сделали - и на сях, и на ноде, и клиенты, и серверы... А маленькую мигалку светодиодом на SST не можете? Жаль. Это говорит не в пользу SST:)

Впрочем, в любом случае - спасибо вам за информацию, я кое-что полезное для себя почерпнул.

 

[ViKo 1]

А маленькую мигалку светодиодом на SST не можете?

И чтобы параллельно SPI-флэш память стиралась и записывалась. :rolleyes:

 

[jcxz 242]

Жаль заметил этот тред поздно, но всё-же заинтересовало утверждение автора, что он использует единый стек на все задачи, при этом ещё и эти задачи могут вытеснять друг друга и будто по функциональности такая система не уступает классической RTOS, где каждая задача имеет свой стек.

Что-то не могу представить - как именно автор добился работы на едином стеке???

И при этом есть возможность использования разделяемых ресурсов с блокировками или нет?

 

Я даже не представляю, какой МК понадобился бы, если решить в многопоточном стиле те задачи, которые я на них запросто решаю в асинхронном однопоточном - с одним единственным стеком. Наверное пришлось бы брать простой PC + линукс(или какой-нибудь андроид) и не париться, только рентабельность такой системы будет никакая, равно как и энергопотребление бешеное.

Имхо: реализовать работу с разделяемыми ресурсами на едином стеке - невозможно.

Например: есть задача1, низкого приоритета, есть задача2 - приоритет её выше. Выполнялась задача1, занимала стек в диапазоне 99...90 байт, тут её вытеснила задача2, заняла стек в диапазоне 89...80. Всё хорошо. Теперь задача2 например хочет записать на флешь, а флешь в это время занята стиранием например. В классической RTOS задача2 войдёт в ожидание готовности некоего семафора и управление будет передано менее приоритетной задаче1 (с переключением на её стек), которая будет выполняться до тех пор пока семафор не перейдёт в состояние "готово" и контекст не будет переключен обратно на задачу2 (и её стек).

А что делать в этом случае в системе ТСа, чтобы данные в диапазоне 89...80 не были порушены задачей1, пока задача2 ожидает готовности флешь???

[Сергей Борщ 143]

Например: есть задача1, низкого приоритета, есть задача2 - приоритет её выше. Выполнялась задача1, занимала стек в диапазоне 99...90 байт, тут её вытеснила задача2, заняла стек в диапазоне 89...80. Всё хорошо. Теперь задача2 например хочет записать на флешь, а флешь в это время занята стиранием например.

На этом задача 2 заканчивается, но перед завершением создает задачу 3, которая будет запущена как только флеш освободится. Задача 2 завершена, ее стек освобожден, задача 1 может продолжать свою работу. Когда флеш освободится - будет запущена задача 3, которая будет работать на вершине стека и докончит запись. Каждая задача строится как лямбда-функция, указатель на которую кладется в соотвествующую очередь. Это именно задачи, а потоки складываются из цепочек задач. Я понял так, пусть brag поправит, если я где-то ошибся.

[AlexandrY 3]

На этом задача 2 заканчивается, но перед завершением создает задачу 3, которая будет запущена как только флеш освободится. Задача 2 завершена, ее стек освобожден, задача 1 может продолжать свою работу.

 

А в жизни задача 2 ничего не знает ни о задаче 3, ни об очередях, ни о чем. Это просто тупая FatFS, и драйвера ее лупят в бесконечном цикле флаг готовности.

 

Нормальная RTOS все равно эту задачу вытеснит и включит другую, либо прерывания от периферии приведут к выполнению других более важных задач,

а brag-у придется переписывать FatFS на лямда функции и флаг ему в руки.

 

FatFS создавалась не один год и brag думаю на это же время оставит нас.

 

 

[brag 0]

И чтобы параллельно SPI-флэш память стиралась и записывалась. :rolleyes:

Ok, запросто. Недельки через 3 у меня появится время, сделаю простенький демонстрационный проектик.

 

Что-то не могу представить - как именно автор добился работы на едином стеке???

И при этом есть возможность использования разделяемых ресурсов с блокировками или нет?

Перечитайте ветку с самого начала и посмотрите ссылки, поймете, как я этого добился. А если не поймете что-то, то спрашивайте.

Блокировок в ССТ нет, это неблокирующий стиль и если воткнете блокировку - все приоритеты ниже или равны заблокируются намертво, тут система работает без блокировок. Есть конечно блокировки для защиты доступа к служебным данным, но они очень короткие(максимум десяток тактов процессора), а в некоторых случаях на "хорошем" процессоре блокировок нет вообще.

 

Имхо: реализовать работу с разделяемыми ресурсами на едином стеке - невозможно.

Например: есть задача1, низкого приоритета, есть задача2 - приоритет её выше. Выполнялась задача1, занимала стек в диапазоне 99...90 байт, тут её вытеснила задача2, заняла стек в диапазоне 89...80. Всё хорошо. Теперь задача2 например хочет записать на флешь, а флешь в это время занята стиранием например. В классической RTOS задача2 войдёт в ожидание готовности некоего семафора и управление будет передано менее приоритетной задаче1 (с переключением на её стек), которая будет выполняться до тех пор пока семафор не перейдёт в состояние "готово" и контекст не будет переключен обратно на задачу2 (и её стек).

А что делать в этом случае в системе ТСа, чтобы данные в диапазоне 89...80 не были порушены задачей1, пока задача2 ожидает готовности флешь???

Перечитайте тему очень внимательно с самого начала и изучите все предоставленные мной исходники, тогда поймете не только, что это возможно, но и поймете как это работает.

В кратце - исполнение кода никогда не блокируется, код всегда отрабатывает до конца, как обычная функция(или стек функций). Но выполнение этого кода в любой момент может быть прервано(вытеснено) другим более приоритетным кодом(будь то прерыванием или любым другим событием). Если нужна сериализация доступа к ресурсу(будь то флешка или что либо другое)- используется не огромный стек и тяжеленные мютексы, а быстрые легковесные очереди(FIFO), часто lock-free. Стек на весь проект один единственный.

 

А в жизни задача 2 ничего не знает ни о задаче 3, ни об очередях, ни о чем. Это просто тупая FatFS, и драйвера ее лупят в бесконечном цикле флаг готовности.

Нужно выкинуть этот тупой цикл и вместо него навесить обработчик события. код станет проще, быстрее и лучше.

 

FatFS создавалась не один год и brag думаю на это же время оставит нас. biggrin.gif

brag, если ему будет нужно - эту FatFS портирует на SST за 1-2 суток, а если FATfs слишком зависима от RTOS, тогда brag отправит ее на мусорку и возьмет(или напишет) более совершенный код, который сможет работать как на RTOS, так и на ССТ, так и вообще без всякой ОС, тоже за несколько суток.

[Сергей Борщ 143]

А в жизни задача 2 ничего не знает ни о задаче 3, ни об очередях, ни о чем. Это просто тупая FatFS

Вы ничего не поняли, но снова все безапелляционно смешали в кучу. Задачу 2, задачу 3, все задачи вы пишете сами.

 

[AlexandrY 3]

brag, если ему будет нужно - эту FatFS портирует на SST за 1-2 суток, а если FATfs слишком зависима от RTOS, тогда brag отправит ее на мусорку и возьмет(или напишет) более совершенный код, который сможет работать как на RTOS, так и на ССТ, так и вообще без всякой ОС, тоже за несколько суток.

 

Воо..! Займитесь.

[jcxz 242]

На этом задача 2 заканчивается, но перед завершением создает задачу 3, которая будет запущена как только флеш освободится. Задача 2 завершена, ее стек освобожден, задача 1 может продолжать свою работу. Когда флеш освободится - будет запущена задача 3, которая будет работать на вершине стека и докончит запись. Каждая задача строится как лямбда-функция, указатель на которую кладется в соотвествующую очередь.

Т.е. - как это задача2 будет завершена? А её куча локальных переменных, дерево вызовов процедур (запись во флешь была вызвана на некоем довольно глубоком уровне вложенности функций) - куда это всё? Где это всё хранить? Как это всё передать задаче3? Если не на стеке - то это крайне неудобно.