message_queue.cpp 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429
  1. /*************************************************************************/
  2. /* message_queue.cpp */
  3. /*************************************************************************/
  4. /* This file is part of: */
  5. /* GODOT ENGINE */
  6. /* http://www.godotengine.org */
  7. /*************************************************************************/
  8. /* Copyright (c) 2007-2016 Juan Linietsky, Ariel Manzur. */
  9. /* */
  10. /* Permission is hereby granted, free of charge, to any person obtaining */
  11. /* a copy of this software and associated documentation files (the */
  12. /* "Software"), to deal in the Software without restriction, including */
  13. /* without limitation the rights to use, copy, modify, merge, publish, */
  14. /* distribute, sublicense, and/or sell copies of the Software, and to */
  15. /* permit persons to whom the Software is furnished to do so, subject to */
  16. /* the following conditions: */
  17. /* */
  18. /* The above copyright notice and this permission notice shall be */
  19. /* included in all copies or substantial portions of the Software. */
  20. /* */
  21. /* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, */
  22. /* EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF */
  23. /* MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.*/
  24. /* IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY */
  25. /* CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, */
  26. /* TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE */
  27. /* SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. */
  28. /*************************************************************************/
  29. #include "message_queue.h"
  30. #include "globals.h"
  31. #include "script_language.h"
  32. MessageQueue *MessageQueue::singleton=NULL;
  33. MessageQueue *MessageQueue::get_singleton() {
  34. return singleton;
  35. }
  36. Error MessageQueue::push_call(ObjectID p_id,const StringName& p_method,const Variant** p_args,int p_argcount,bool p_show_error) {
  37. _THREAD_SAFE_METHOD_
  38. int room_needed=sizeof(Message)+sizeof(Variant)*p_argcount;
  39. if ((buffer_end+room_needed) >= buffer_size) {
  40. String type;
  41. if (ObjectDB::get_instance(p_id))
  42. type=ObjectDB::get_instance(p_id)->get_type();
  43. print_line("failed method: "+type+":"+p_method+" target ID: "+itos(p_id));
  44. statistics();
  45. }
  46. ERR_FAIL_COND_V( (buffer_end+room_needed) >= buffer_size , ERR_OUT_OF_MEMORY );
  47. Message * msg = memnew_placement( &buffer[ buffer_end ], Message );
  48. msg->args=p_argcount;
  49. msg->instance_ID=p_id;
  50. msg->target=p_method;
  51. msg->type=TYPE_CALL;
  52. if (p_show_error)
  53. msg->type|=FLAG_SHOW_ERROR;
  54. buffer_end+=sizeof(Message);
  55. for(int i=0;i<p_argcount;i++) {
  56. Variant * v = memnew_placement( &buffer[ buffer_end ], Variant );
  57. buffer_end+=sizeof(Variant);
  58. *v=*p_args[i];
  59. }
  60. return OK;
  61. }
  62. Error MessageQueue::push_call(ObjectID p_id, const StringName& p_method, VARIANT_ARG_DECLARE) {
  63. VARIANT_ARGPTRS;
  64. int argc=0;
  65. for(int i=0;i<VARIANT_ARG_MAX;i++) {
  66. if (argptr[i]->get_type()==Variant::NIL)
  67. break;
  68. argc++;
  69. }
  70. return push_call(p_id,p_method,argptr,argc,false);
  71. }
  72. Error MessageQueue::push_set(ObjectID p_id, const StringName& p_prop, const Variant& p_value) {
  73. _THREAD_SAFE_METHOD_
  74. uint8_t room_needed=sizeof(Message)+sizeof(Variant);
  75. if ((buffer_end+room_needed) >= buffer_size) {
  76. String type;
  77. if (ObjectDB::get_instance(p_id))
  78. type=ObjectDB::get_instance(p_id)->get_type();
  79. print_line("failed set: "+type+":"+p_prop+" target ID: "+itos(p_id));
  80. statistics();
  81. }
  82. ERR_FAIL_COND_V( (buffer_end+room_needed) >= buffer_size , ERR_OUT_OF_MEMORY );
  83. Message * msg = memnew_placement( &buffer[ buffer_end ], Message );
  84. msg->args=1;
  85. msg->instance_ID=p_id;
  86. msg->target=p_prop;
  87. msg->type=TYPE_SET;
  88. buffer_end+=sizeof(Message);
  89. Variant * v = memnew_placement( &buffer[ buffer_end ], Variant );
  90. buffer_end+=sizeof(Variant);
  91. *v=p_value;
  92. return OK;
  93. }
  94. Error MessageQueue::push_notification(ObjectID p_id, int p_notification) {
  95. _THREAD_SAFE_METHOD_
  96. ERR_FAIL_COND_V(p_notification<0, ERR_INVALID_PARAMETER );
  97. uint8_t room_needed=sizeof(Message);
  98. if ((buffer_end+room_needed) >= buffer_size) {
  99. String type;
  100. if (ObjectDB::get_instance(p_id))
  101. type=ObjectDB::get_instance(p_id)->get_type();
  102. print_line("failed notification: "+itos(p_notification)+" target ID: "+itos(p_id));
  103. statistics();
  104. }
  105. ERR_FAIL_COND_V( (buffer_end+room_needed) >= buffer_size , ERR_OUT_OF_MEMORY );
  106. Message * msg = memnew_placement( &buffer[ buffer_end ], Message );
  107. msg->type=TYPE_NOTIFICATION;
  108. msg->instance_ID=p_id;
  109. //msg->target;
  110. msg->notification=p_notification;
  111. buffer_end+=sizeof(Message);
  112. return OK;
  113. }
  114. Error MessageQueue::push_call(Object *p_object, const StringName& p_method, VARIANT_ARG_DECLARE) {
  115. return push_call(p_object->get_instance_ID(),p_method,VARIANT_ARG_PASS);
  116. }
  117. Error MessageQueue::push_notification(Object *p_object, int p_notification) {
  118. return push_notification(p_object->get_instance_ID(),p_notification);
  119. }
  120. Error MessageQueue::push_set(Object *p_object, const StringName& p_prop, const Variant& p_value) {
  121. return push_set(p_object->get_instance_ID(),p_prop,p_value);
  122. }
  123. void MessageQueue::statistics() {
  124. Map<StringName,int> set_count;
  125. Map<int,int> notify_count;
  126. Map<StringName,int> call_count;
  127. int null_count=0;
  128. uint32_t read_pos=0;
  129. while (read_pos < buffer_end ) {
  130. Message *message = (Message*)&buffer[ read_pos ];
  131. Object *target = ObjectDB::get_instance(message->instance_ID);
  132. if (target!=NULL) {
  133. switch(message->type&FLAG_MASK) {
  134. case TYPE_CALL: {
  135. if (!call_count.has(message->target))
  136. call_count[message->target]=0;
  137. call_count[message->target]++;
  138. } break;
  139. case TYPE_NOTIFICATION: {
  140. if (!notify_count.has(message->notification))
  141. notify_count[message->notification]=0;
  142. notify_count[message->notification]++;
  143. } break;
  144. case TYPE_SET: {
  145. if (!set_count.has(message->target))
  146. set_count[message->target]=0;
  147. set_count[message->target]++;
  148. } break;
  149. }
  150. //object was deleted
  151. //WARN_PRINT("Object was deleted while awaiting a callback")
  152. //should it print a warning?
  153. } else {
  154. null_count++;
  155. }
  156. read_pos+=sizeof(Message);
  157. if ((message->type&FLAG_MASK)!=TYPE_NOTIFICATION)
  158. read_pos+=sizeof(Variant)*message->args;
  159. }
  160. print_line("TOTAL BYTES: "+itos(buffer_end));
  161. print_line("NULL count: "+itos(null_count));
  162. for(Map<StringName,int>::Element *E=set_count.front();E;E=E->next()) {
  163. print_line("SET "+E->key()+": "+itos(E->get()));
  164. }
  165. for(Map<StringName,int>::Element *E=call_count.front();E;E=E->next()) {
  166. print_line("CALL "+E->key()+": "+itos(E->get()));
  167. }
  168. for(Map<int,int>::Element *E=notify_count.front();E;E=E->next()) {
  169. print_line("NOTIFY "+itos(E->key())+": "+itos(E->get()));
  170. }
  171. }
  172. bool MessageQueue::print() {
  173. #if 0
  174. uint32_t read_pos=0;
  175. while (read_pos < buffer_end ) {
  176. Message *message = (Message*)&buffer[ read_pos ];
  177. Object *target = ObjectDB::get_instance(message->instance_ID);
  178. String cname;
  179. String cfunc;
  180. if (target==NULL) {
  181. //object was deleted
  182. //WARN_PRINT("Object was deleted while awaiting a callback")
  183. //should it print a warning?
  184. } else if (message->notification>=0) {
  185. // messages don't expect a return value
  186. cfunc="notification # "+itos(message->notification);
  187. cname=target->get_type();
  188. } else if (!message->target.empty()) {
  189. cfunc="property: "+message->target;
  190. cname=target->get_type();
  191. } else if (message->target) {
  192. cfunc=String(message->target)+"()";
  193. cname=target->get_type();
  194. }
  195. read_pos+=sizeof(Message);
  196. if (message->type!=TYPE_NOTIFICATION)
  197. read_pos+=sizeof(Variant)*message->args;
  198. }
  199. #endif
  200. return false;
  201. }
  202. int MessageQueue::get_max_buffer_usage() const {
  203. return buffer_max_used;
  204. }
  205. void MessageQueue::_call_function(Object* p_target, const StringName& p_func, const Variant *p_args, int p_argcount,bool p_show_error) {
  206. const Variant **argptrs=NULL;
  207. if (p_argcount) {
  208. argptrs = (const Variant**)alloca(sizeof(Variant*)*p_argcount);
  209. for(int i=0;i<p_argcount;i++) {
  210. argptrs[i]=&p_args[i];
  211. }
  212. }
  213. Variant::CallError ce;
  214. p_target->call(p_func,argptrs,p_argcount,ce);
  215. if (p_show_error && ce.error!=Variant::CallError::CALL_OK) {
  216. ERR_PRINTS("Error calling deferred method: "+Variant::get_call_error_text(p_target,p_func,argptrs,p_argcount,ce));
  217. }
  218. }
  219. void MessageQueue::flush() {
  220. if (buffer_end > buffer_max_used) {
  221. buffer_max_used=buffer_end;
  222. //statistics();
  223. }
  224. uint32_t read_pos=0;
  225. //using reverse locking strategy
  226. _THREAD_SAFE_LOCK_
  227. while (read_pos<buffer_end) {
  228. _THREAD_SAFE_UNLOCK_
  229. //lock on each interation, so a call can re-add itself to the message queue
  230. Message *message = (Message*)&buffer[ read_pos ];
  231. Object *target = ObjectDB::get_instance(message->instance_ID);
  232. if (target!=NULL) {
  233. switch(message->type&FLAG_MASK) {
  234. case TYPE_CALL: {
  235. Variant *args= (Variant*)(message+1);
  236. // messages don't expect a return value
  237. _call_function(target,message->target,args,message->args,message->type&FLAG_SHOW_ERROR);
  238. for(int i=0;i<message->args;i++) {
  239. args[i].~Variant();
  240. }
  241. } break;
  242. case TYPE_NOTIFICATION: {
  243. // messages don't expect a return value
  244. target->notification(message->notification);
  245. } break;
  246. case TYPE_SET: {
  247. Variant *arg= (Variant*)(message+1);
  248. // messages don't expect a return value
  249. target->set(message->target,*arg);
  250. arg->~Variant();
  251. } break;
  252. }
  253. }
  254. uint32_t advance = sizeof(Message);
  255. if ((message->type&FLAG_MASK)!=TYPE_NOTIFICATION)
  256. advance+=sizeof(Variant)*message->args;
  257. message->~Message();
  258. _THREAD_SAFE_LOCK_
  259. read_pos+=advance;
  260. }
  261. buffer_end=0; // reset buffer
  262. _THREAD_SAFE_UNLOCK_
  263. }
  264. MessageQueue::MessageQueue() {
  265. ERR_FAIL_COND(singleton!=NULL);
  266. singleton=this;
  267. buffer_end=0;
  268. buffer_max_used=0;
  269. buffer_size=GLOBAL_DEF( "core/message_queue_size_kb", DEFAULT_QUEUE_SIZE_KB );
  270. buffer_size*=1024;
  271. buffer = memnew_arr( uint8_t, buffer_size );
  272. }
  273. MessageQueue::~MessageQueue() {
  274. uint32_t read_pos=0;
  275. while (read_pos < buffer_end ) {
  276. Message *message = (Message*)&buffer[ read_pos ];
  277. Variant *args= (Variant*)(message+1);
  278. int argc = message->args;
  279. if ((message->type&FLAG_MASK)!=TYPE_NOTIFICATION) {
  280. for (int i=0;i<argc;i++)
  281. args[i].~Variant();
  282. }
  283. message->~Message();
  284. read_pos+=sizeof(Message);
  285. if ((message->type&FLAG_MASK)!=TYPE_NOTIFICATION)
  286. read_pos+=sizeof(Variant)*message->args;
  287. }
  288. singleton=NULL;
  289. memdelete_arr( buffer );
  290. }