POSIX IPC-Message Queue in Linux

یکی از تکنیکهایی که در دنیا برنامه نویسی از اهمیت زیادی برخوردار می باشد، مبحث ارتباطات بین پردازشی در هسته سیستم عامل می باشد

در سیستم هایی که از حساسیت بالایی برخوردار هستند، مانند سیستم های حیاتی ایمنی که سرعت اجرای کد های نوشته شده بر حسب میکرو ثانیه و نانو ثانیه اندازه گیری می شوند بازدهی سیستم و همچنین مصرف منابع سخت افزاری کمتر بسیار مهم می باشد. معمولا در سیستم های ایمنی حیاتی به لحاظ تضمین عملکرد سیستم از الگوریتمهای

Voted (get 2 of 3)

استفاده می شود، به این معنی که همزمان در تایم های یکسان باید سه نود محاسباتی وجود داشته باشد که هریک در زمانهای مشخص داده های حساس را به رای قرار بدهند تا بدین ترتیب بتوان رفتارهای سامانه را تضمین کرد بنابراین با توضیحات کوتاهی که ذکر شد، به یکی از استانداردهای پازیکس در سیستم عامل لینوکس خواهیم پرداخت که جهت به اشتراک گذاری دادها بین چند پردازه در سطح هسته سیستم عامل کاربرد دارند، البته سناریو پیش روی ما فقط یکی از تکنیک های ارتباطات بین پردازه ای در لینوکس می باشد روشهای دیگری هم مانند حافظه اشتراکی ، پایپنگ ، سوکت های محلی ، سوکت های شبکه و ... وجود دارند

تصور کنید دو پردازه یکسان دارید که می خواهید با استفاده از تکنیک صف های پیام دادهایی را دریافت کنید و یا یک عملکرد مشخص را در زمان دریافت پیام انجام دهید، توجه داشته باشید که با استفاده از این روش می توانید برنامه هایی تدارک ببینید که رویداد محور باشند

Event Handler

دقت داشته باشید که ما در این سناریو از هر سه زبان سی و سی پلاس پلاس و اسمبلی جهت پیاده سازی این سناریو استفاده کرده ایم

ابتدا یک کلاس جهت مدیریت فعال کردن یک صف پیام وهمچنین کنترل کننده ارسال و دریافت پیامها به شکل زیر تعریف میکنیم

به علت ناسازگاری لیندکدین با یونیکد سورس کد ها این آموزش در آدرس زیر قرار گرفته است

#include <unistd.h>
#include <signal.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <mqueue.h>
#include <vector>


std::string kQueueName = "/test";
struct Message {
	int x , y ;
};


template<class MsgType>
class MessageQueue {
private:
	mqd_t m_qhandle;
    typedef struct mq_attr qattr;
    qattr* m_attrs;


    void setAttribute(int max_count , int max_size){
    	mq_getattr(m_qhandle , m_attrs);
    	(*m_attrs).mq_maxmsg  = max_count;/* Maximum number of messages.  */
    	(*m_attrs).mq_msgsize = max_size;	/* Maximum message size.  */
    	//(*m_attrs).mq_flags   = 0;        	/* Message queue flags.  */
    	//mq_setattr(m_qhandle , m_attrs , m_attrs);
    }
public:
	MessageQueue() {}
	~MessageQueue(){
		mq_close(m_qhandle);
	}


	void operator()(const std::string& name , int flags , int max_count =10 , int max_size=sizeof(MsgType)){
		m_qhandle = mq_open(name.c_str() , flags | O_CREAT , 0666 , m_attrs);
		if(m_qhandle < 0){
			throw std::runtime_error("Failed to open a queue");
		}
		setAttribute(max_count , max_size);
	}
	void Send(const char* data){
		if(mq_send(m_qhandle , data , (*m_attrs).mq_msgsize ,0) < 0){
			throw std::runtime_error("Failed to send message");
		}
	}


	void Receive(char* data){
		if(mq_receive(m_qhandle , data , (*m_attrs).mq_msgsize , 0) < 0){
			throw std::runtime_error("Failed to receive message");
		}
	}

};

و بعد از تعریف کلاس فوق باید یک کلاس ارسال کننده اطلاعات بر روی صف تدارک ببینیم که به صورت زیر تعریف شده است

template<class MsgType>
	class Writer {
	private:
	  typedef std::vector<MsgType> vecType;
		MessageQueue<MsgType> m_queue;
		void writeMessageQ(const MsgType& data){
			m_queue.Send(reinterpret_cast<const char*>(&data));
		}
	public:
		Writer(std::string& name)  { m_queue(name,O_WRONLY); }
	

	  template<typename ITR>
		void setMessage(vecType& messages , ITR& itr){
			for(; itr != messages.cend() ; itr++){
				std::cout << "Write Message " << itr->x << " - " << itr->y << '\n';
				writeMessageQ(*itr);
				usleep(100000);
			}
		}
	  void setMessage(){
			vecType messages = {{1,0} , {0 , 1}, {1 , 1} ,{0 , 0}};
			setMessage(messages , messages.cbegin()); 
	  }
	  
	};

همانطور که مشخص هست، کلاس فوق یک وکتور از نوع داده ای پیامی که جهت ارسال تدارک دیدم را ایجاد می کند و با استفاده از کلاس صف دادها را در صف وارد می نماید بعد از تعریف کلاس فوق نیاز به یک کلاس دریافت کننده نیز می باشد که بتواند داده های را از صف استخراج نماید بنابراین به شرح ذیل کلاس دریافت کننده را تعریف خواهیم کرد

template<class MsgType ,
         class Strategy = MessageStrategy >
class Reader {
private:
	typedef void(Strategy::*funcPtr_t)(const MsgType&);
	MessageQueue<MsgType> m_queue;
	funcPtr_t m_callBack;
public:
	Reader(std::string& name ,funcPtr_t funcPtr = &Strategy::callBack ) :
		 m_callBack(funcPtr) { m_queue(name,O_RDONLY); }


	void getMessage() {
		MsgType data;
		Strategy m_strategy;
	     while(true){
	    	 m_queue.Receive(reinterpret_cast<char*>(&data));
	    	 (m_strategy.*m_callBack)(data);
	     }
	}


protected:


};

توجه داشته باشید که کلاس مدیریت کننده صف ها باید در هر دو پردازه وجود داشته باشد، ولی همانطور که مشخص هست کلاس ارسال کننده و دریافت کننده باید در پردازه های متفاوت از هم وجود داشته باشند و به صورت جداگانه کامپایل شوند ویا در صورتی که که مایل باشید می توانید از کو روتین هایی مانند فورک که جز توابع سطح پایین هسته سیستم عامل لینوکس هستند به شکل زیر استفاده نمایید

    pid_t tpid = fork();
    if(tpid){
	    Writer<Message> writer(kQueueName);
	    writer.setMessage();
    	usleep(100000);
    	kill(tpid , SIGTERM);
    }else {
	   Reader<Message> reader(kQueueName);
	   reader.getMessage();
	   mq_unlink(kQueueName.c_str());
	   
    }

همانطور که مشخص هست طبق عملکرد تابع فورک پردازه اصلی که با شماره ای بزرگتر از 0 ایجاد می شود دارای یک پردازه فرزند می باشد که با یک شماره 0 مشخص و اجرا خواهند شد البته در دوفضای آدرس دهی متفاوت از هم بنابراین ابتدا کلاس ارسال کننده اجرا خواهد شد و در پردازه فرزند هم کلاس دریافت کننده اجرا خواهد شد همانطور که عرض شد می توانید کلاسها را جدا گانه از هم نیز کامپایل کنید به علت اینکه شرح عملکرد توابع هسته سیستم عامل که برای اجرای پردازه ها در لینوکس استفاده می شود معمولا پیچیدگی های زیادی دارند

نکته ای که درباره کلاس دریافت کننده وجود دارد این موضوع می باشد که می توانید منطق عملکردی خود را به صورت یک الگوی استراتژی تعریف کنید و همانطور هم که مشاهده می کنید کلاس دریافت کننده در حلقه ای که جهت دریافت اطلاعات تدارک دیده است این اطلاعات را برای کلاس استراتژی ما ارسال خواهد کرد بنابراین می توانید چنین کلاسی راهم جهت بارگذاری منطق استراتژی برنامه در زمان دریافت کردن داده تعریف نمایید

extern "C" void __attribute__((cdecl)) fixUpMessage(const void* );


class MessageStrategy {
public:
	template<class MsgType>
	void callBack(const MsgType& data){
		fixUpMessage(&data);
		std::cout << "Recived Message From handle Num( "<< "): " << data.x << " - "  << data.y << '\n';
	}
	
};

بنابراین با استفاده از کلاس استراتژی فوق می توانیم عملکردهای متفاوتی را هم به برنامه تزریق نماییم نکته ای هم که در کلاس فوق مشخص هست این موضوع هست که معمولا از تکنیک های ارتباطات بین پردازه ای در سیستم هایی استفاده می شود که قسمت های حیاتی که با داده های حساس در ارتباط هستند بسیار مهم می باشد بنابراین این قسمت ها را هم میتوانید به صورت مستقیم در اسمبلی بنویسید دقیقا کاری که من در این سناریو انجام داده ام تابعی را مستقیما در اسمبلی نوشته ام و در این سناریو از آن استفاده کرده ام که می تواند به شکل زیر باشد نکته ای که مهم است توجه داشته باشید، به این علت که در این کلاس قصد داشتیم یکپارچگی نوع را در زمان اجرا داشته باشیم نوع پارامتر ورودی تابع اکسترن که در زمان کامپایل به تابع اسمبلی پیونده زده میشود را از نوع

void*

تعریف کرده ام، که هر نوع فرمتی که پیامها داشتند را بتوانید ارسال کنید البته با این کار ملاحظات بیشتری لازم هست که داشته باشید تعریف این نوع هم به علت پشتیبانی نشدن نوع های تمپلیت در زمان تعریف توابع اکسترن می باشد

.globl fixUpMessage
.type fixUpMessage, @function


Offset_Y = 0x4


.section .data
    MsgBase:
              .long 0                 # offset of to message structure


.section .bss


.section .text
fixUpMessage:
.cfi_startproc
     push     %ebp
     mov      %esp      , %ebp


     lea      (MsgBase) , %esi        # store offset MsgBase into esi
     mov      0x8(%ebp) , %eax        # store first param function (Message*)
     movl     %eax      , (%esi)      # store offset param into MsgBase
     mov      (MsgBase) , %eax        # get the content of MsgBase into eax
     add      $0x1e     , (%eax)      # equ Message.x + 5
     add      $0x1f,Offset_Y(%eax)    # equ Message.y + 6


     mov      %ebp 		, %esp
     pop      %ebp
     ret
.cfi_endproc

توجه داشته باشید که در زمان کامپایل حتما باید کتابخانه

rt (Linux Real Time libraray)

را هم برای لینکر مشخص نمایید، و زمانی که باینری فایلها را اجرا نمایید اگر به صورت جدا جدا کامپایل کرده باشید برنامه دریافت کننده منتظر دریافت پیام در یک حلقه بی نهایت می ماند ولی اگر از فورک استفاده کرده باشید بعد از هر بار ارسال داده در برنامه دریافت کننده داده مذکور را برای تابع اسمبلی ارسال خواهد کرد و نتیجه را چاپ خواهد کرد

توجه داشته باشید که به علت پیچیدگی که به وجود می آمد از تکنیک های همزمانی و نخ های ایمن استفاده نشده است در یک آموزش دیگر به بحث

Thread Saftey in Message Queue

خواهیم پرداخت، همانطور که مشخص هست استفاده از صف های پیام در خیلی از سناریو ها کاربرد دارد منتهی لازم هست که کمی با کرنل لینوکس و نحوه مدیرت پردازش ها و البته مباحث

Inter Process Communication

در سیستم عامل لینوکس اطلاعات فنی داشته باشید تا هرچه بهتر بتوانید از این تکنیک ها در جهت هرچه بهتر شدن نرم افزارهای خود استفاده نمایید

موفق باشید

در صورتی که ابهامی در بخشی از این آموزش وجود داشت، در کامنت ها مرقوم بفرمایید