EQStreamFactory.cpp 10 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421
  1. /*
  2. EQ2Emulator: Everquest II Server Emulator
  3. Copyright (C) 2007 EQ2EMulator Development Team (http://www.eq2emulator.net)
  4. This file is part of EQ2Emulator.
  5. EQ2Emulator is free software: you can redistribute it and/or modify
  6. it under the terms of the GNU General Public License as published by
  7. the Free Software Foundation, either version 3 of the License, or
  8. (at your option) any later version.
  9. EQ2Emulator is distributed in the hope that it will be useful,
  10. but WITHOUT ANY WARRANTY; without even the implied warranty of
  11. MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  12. GNU General Public License for more details.
  13. You should have received a copy of the GNU General Public License
  14. along with EQ2Emulator. If not, see <http://www.gnu.org/licenses/>.
  15. */
  16. #include "EQStreamFactory.h"
  17. #include "Log.h"
  18. #ifdef WIN32
  19. #include <WinSock2.h>
  20. #include <windows.h>
  21. #include <process.h>
  22. #include <io.h>
  23. #include <stdio.h>
  24. #else
  25. #include <sys/socket.h>
  26. #include <netinet/in.h>
  27. #include <sys/select.h>
  28. #include <arpa/inet.h>
  29. #include <netdb.h>
  30. #include <pthread.h>
  31. #endif
  32. #include <fcntl.h>
  33. #include <fstream>
  34. #include <iostream>
  35. #include "op_codes.h"
  36. #include "EQStream.h"
  37. #include "packet_dump.h"
  38. #ifdef WORLD
  39. #include "../WorldServer/client.h"
  40. #endif
  41. using namespace std;
  42. #ifdef WORLD
  43. extern ClientList client_list;
  44. #endif
  45. ThreadReturnType EQStreamFactoryReaderLoop(void *eqfs)
  46. {
  47. if(eqfs){
  48. EQStreamFactory *fs=(EQStreamFactory *)eqfs;
  49. fs->ReaderLoop();
  50. }
  51. THREAD_RETURN(NULL);
  52. }
  53. ThreadReturnType EQStreamFactoryWriterLoop(void *eqfs)
  54. {
  55. if(eqfs){
  56. EQStreamFactory *fs=(EQStreamFactory *)eqfs;
  57. fs->WriterLoop();
  58. }
  59. THREAD_RETURN(NULL);
  60. }
  61. ThreadReturnType EQStreamFactoryCombinePacketLoop(void *eqfs)
  62. {
  63. if(eqfs){
  64. EQStreamFactory *fs=(EQStreamFactory *)eqfs;
  65. fs->CombinePacketLoop();
  66. }
  67. THREAD_RETURN(NULL);
  68. }
  69. EQStreamFactory::EQStreamFactory(EQStreamType type, int port)
  70. {
  71. StreamType=type;
  72. Port=port;
  73. listen_ip_address = 0;
  74. }
  75. void EQStreamFactory::Close()
  76. {
  77. CheckTimeout(true);
  78. Stop();
  79. #ifdef WIN32
  80. closesocket(sock);
  81. #else
  82. close(sock);
  83. #endif
  84. sock=-1;
  85. }
  86. bool EQStreamFactory::Open()
  87. {
  88. struct sockaddr_in address;
  89. #ifndef WIN32
  90. pthread_t t1, t2, t3;
  91. #endif
  92. /* Setup internet address information.
  93. This is used with the bind() call */
  94. memset((char *) &address, 0, sizeof(address));
  95. address.sin_family = AF_INET;
  96. address.sin_port = htons(Port);
  97. #if defined(LOGIN) || defined(MINILOGIN)
  98. if(listen_ip_address)
  99. address.sin_addr.s_addr = inet_addr(listen_ip_address);
  100. else
  101. address.sin_addr.s_addr = htonl(INADDR_ANY);
  102. #else
  103. address.sin_addr.s_addr = htonl(INADDR_ANY);
  104. #endif
  105. /* Setting up UDP port for new clients */
  106. sock = socket(AF_INET, SOCK_DGRAM, 0);
  107. if (sock < 0) {
  108. return false;
  109. }
  110. if (bind(sock, (struct sockaddr *) &address, sizeof(address)) < 0) {
  111. close(sock);
  112. sock=-1;
  113. return false;
  114. }
  115. #ifdef WIN32
  116. unsigned long nonblock = 1;
  117. ioctlsocket(sock, FIONBIO, &nonblock);
  118. #else
  119. fcntl(sock, F_SETFL, O_NONBLOCK);
  120. #endif
  121. //moved these because on windows the output was delayed and causing the console window to look bad
  122. LogWrite(WORLD__DEBUG, 0, "World", "Starting factory Reader");
  123. LogWrite(WORLD__DEBUG, 0, "World", "Starting factory Writer");
  124. #ifdef WIN32
  125. _beginthread(EQStreamFactoryReaderLoop,0, this);
  126. _beginthread(EQStreamFactoryWriterLoop,0, this);
  127. _beginthread(EQStreamFactoryCombinePacketLoop,0, this);
  128. #else
  129. pthread_create(&t1,NULL,EQStreamFactoryReaderLoop,this);
  130. pthread_create(&t2,NULL,EQStreamFactoryWriterLoop,this);
  131. pthread_create(&t3,NULL,EQStreamFactoryCombinePacketLoop,this);
  132. pthread_detach(t1);
  133. pthread_detach(t2);
  134. pthread_detach(t3);
  135. #endif
  136. return true;
  137. }
  138. EQStream *EQStreamFactory::Pop()
  139. {
  140. EQStream *s=NULL;
  141. //cout << "Pop():Locking MNewStreams" << endl;
  142. MNewStreams.lock();
  143. if (NewStreams.size()) {
  144. s=NewStreams.front();
  145. NewStreams.pop();
  146. s->PutInUse();
  147. }
  148. MNewStreams.unlock();
  149. //cout << "Pop(): Unlocking MNewStreams" << endl;
  150. return s;
  151. }
  152. void EQStreamFactory::Push(EQStream *s)
  153. {
  154. //cout << "Push():Locking MNewStreams" << endl;
  155. MNewStreams.lock();
  156. NewStreams.push(s);
  157. MNewStreams.unlock();
  158. //cout << "Push(): Unlocking MNewStreams" << endl;
  159. }
  160. void EQStreamFactory::ReaderLoop()
  161. {
  162. fd_set readset;
  163. map<string,EQStream *>::iterator stream_itr;
  164. int num;
  165. int length;
  166. unsigned char buffer[2048];
  167. sockaddr_in from;
  168. int socklen=sizeof(sockaddr_in);
  169. timeval sleep_time;
  170. ReaderRunning=true;
  171. while(sock!=-1) {
  172. MReaderRunning.lock();
  173. if (!ReaderRunning)
  174. break;
  175. MReaderRunning.unlock();
  176. FD_ZERO(&readset);
  177. FD_SET(sock,&readset);
  178. sleep_time.tv_sec=30;
  179. sleep_time.tv_usec=0;
  180. if ((num=select(sock+1,&readset,NULL,NULL,&sleep_time))<0) {
  181. // What do we wanna do?
  182. } else if (num==0)
  183. continue;
  184. if (FD_ISSET(sock,&readset)) {
  185. #ifdef WIN32
  186. if ((length=recvfrom(sock,(char*)buffer,sizeof(buffer),0,(struct sockaddr*)&from,(int *)&socklen))<0)
  187. #else
  188. if ((length=recvfrom(sock,buffer,2048,0,(struct sockaddr *)&from,(socklen_t *)&socklen))<0)
  189. #endif
  190. {
  191. // What do we wanna do?
  192. } else {
  193. char temp[25];
  194. sprintf(temp,"%u.%d",ntohl(from.sin_addr.s_addr),ntohs(from.sin_port));
  195. MStreams.lock();
  196. if ((stream_itr=Streams.find(temp))==Streams.end() || buffer[1]==OP_SessionRequest) {
  197. MStreams.unlock();
  198. if (buffer[1]==OP_SessionRequest) {
  199. if(stream_itr != Streams.end() && stream_itr->second)
  200. stream_itr->second->SetState(CLOSED);
  201. EQStream *s=new EQStream(from);
  202. s->SetFactory(this);
  203. s->SetStreamType(StreamType);
  204. Streams[temp]=s;
  205. WriterWork.Signal();
  206. Push(s);
  207. s->Process(buffer,length);
  208. s->SetLastPacketTime(Timer::GetCurrentTime2());
  209. }
  210. } else {
  211. EQStream *curstream = stream_itr->second;
  212. //dont bother processing incoming packets for closed connections
  213. if(curstream->CheckClosed())
  214. curstream = NULL;
  215. else
  216. curstream->PutInUse();
  217. MStreams.unlock();
  218. if(curstream) {
  219. curstream->Process(buffer,length);
  220. curstream->SetLastPacketTime(Timer::GetCurrentTime2());
  221. curstream->ReleaseFromUse();
  222. }
  223. }
  224. }
  225. }
  226. }
  227. }
  228. void EQStreamFactory::CheckTimeout(bool remove_all)
  229. {
  230. //lock streams the entire time were checking timeouts, it should be fast.
  231. MStreams.lock();
  232. unsigned long now=Timer::GetCurrentTime2();
  233. map<string,EQStream *>::iterator stream_itr;
  234. for(stream_itr=Streams.begin();stream_itr!=Streams.end();) {
  235. EQStream *s = stream_itr->second;
  236. EQStreamState state = s->GetState();
  237. if (state==CLOSING && !s->HasOutgoingData()) {
  238. stream_itr->second->SetState(CLOSED);
  239. state = CLOSED;
  240. } else if (s->CheckTimeout(now, STREAM_TIMEOUT)) {
  241. const char* stateString;
  242. switch (state){
  243. case ESTABLISHED:
  244. stateString = "Established";
  245. break;
  246. case CLOSING:
  247. stateString = "Closing";
  248. break;
  249. case CLOSED:
  250. stateString = "Closed";
  251. break;
  252. default:
  253. stateString = "Unknown";
  254. break;
  255. }
  256. LogWrite(WORLD__DEBUG, 0, "World", "Timeout up!, state=%s", stateString);
  257. if (state==ESTABLISHED) {
  258. s->Close();
  259. } else if (state == CLOSING) {
  260. //if we time out in the closing state, just give up
  261. s->SetState(CLOSED);
  262. state = CLOSED;
  263. }
  264. }
  265. //not part of the else so we check it right away on state change
  266. if (remove_all || state==CLOSED) {
  267. if (!remove_all && s->getTimeoutDelays()<2) {
  268. s->addTimeoutDelay();
  269. //give it a little time for everybody to finish with it
  270. } else {
  271. //everybody is done, we can delete it now
  272. LogWrite(WORLD__DEBUG, 0, "World", "Removing connection...");
  273. map<string,EQStream *>::iterator temp=stream_itr;
  274. stream_itr++;
  275. //let whoever has the stream outside delete it
  276. #ifdef WORLD
  277. client_list.RemoveConnection(temp->second);
  278. #endif
  279. delete temp->second;
  280. Streams.erase(temp);
  281. continue;
  282. }
  283. }
  284. stream_itr++;
  285. }
  286. MStreams.unlock();
  287. }
  288. void EQStreamFactory::CombinePacketLoop(){
  289. deque<EQStream*> combine_que;
  290. CombinePacketRunning = true;
  291. bool packets_waiting = false;
  292. while(sock!=-1) {
  293. if (!CombinePacketRunning)
  294. break;
  295. MStreams.lock();
  296. map<string,EQStream *>::iterator stream_itr;
  297. for(stream_itr=Streams.begin();stream_itr!=Streams.end();stream_itr++) {
  298. if(!stream_itr->second){
  299. continue;
  300. }
  301. if(stream_itr->second->combine_timer && stream_itr->second->combine_timer->Check())
  302. combine_que.push_back(stream_itr->second);
  303. }
  304. EQStream* stream = 0;
  305. packets_waiting = false;
  306. while(combine_que.size()){
  307. stream = combine_que.front();
  308. if(stream->CheckActive()){
  309. if(!stream->CheckCombineQueue())
  310. packets_waiting = true;
  311. }
  312. combine_que.pop_front();
  313. }
  314. MStreams.unlock();
  315. if(!packets_waiting)
  316. Sleep(25);
  317. Sleep(1);
  318. }
  319. }
  320. void EQStreamFactory::WriterLoop()
  321. {
  322. map<string,EQStream *>::iterator stream_itr;
  323. vector<EQStream *> wants_write;
  324. vector<EQStream *>::iterator cur,end;
  325. deque<EQStream*> resend_que;
  326. bool decay=false;
  327. uint32 stream_count;
  328. Timer DecayTimer(20);
  329. WriterRunning=true;
  330. DecayTimer.Enable();
  331. while(sock!=-1) {
  332. Timer::SetCurrentTime();
  333. //if (!havework) {
  334. //WriterWork.Wait();
  335. //}
  336. MWriterRunning.lock();
  337. if (!WriterRunning)
  338. break;
  339. MWriterRunning.unlock();
  340. wants_write.clear();
  341. decay=DecayTimer.Check();
  342. //copy streams into a seperate list so we dont have to keep
  343. //MStreams locked while we are writting
  344. MStreams.lock();
  345. for(stream_itr=Streams.begin();stream_itr!=Streams.end();stream_itr++) {
  346. // If it's time to decay the bytes sent, then let's do it before we try to write
  347. if(!stream_itr->second){
  348. Streams.erase(stream_itr);
  349. break;
  350. }
  351. if (decay)
  352. stream_itr->second->Decay();
  353. if (stream_itr->second->HasOutgoingData()) {
  354. stream_itr->second->PutInUse();
  355. wants_write.push_back(stream_itr->second);
  356. }
  357. if(stream_itr->second->resend_que_timer->Check())
  358. resend_que.push_back(stream_itr->second);
  359. }
  360. MStreams.unlock();
  361. //do the actual writes
  362. cur = wants_write.begin();
  363. end = wants_write.end();
  364. for(; cur != end; cur++) {
  365. (*cur)->Write(sock);
  366. (*cur)->ReleaseFromUse();
  367. }
  368. while(resend_que.size()){
  369. resend_que.front()->CheckResend(sock);
  370. resend_que.pop_front();
  371. }
  372. Sleep(10);
  373. MStreams.lock();
  374. stream_count=Streams.size();
  375. MStreams.unlock();
  376. if (!stream_count) {
  377. //cout << "No streams, waiting on condition" << endl;
  378. WriterWork.Wait();
  379. //cout << "Awake from condition, must have a stream now" << endl;
  380. }
  381. }
  382. }