9
3

EQStreamFactory.cpp 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434
  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. #ifdef LOGIN
  123. LogWrite(LOGIN__DEBUG, 0, "Login", "Starting factory Reader");
  124. LogWrite(LOGIN__DEBUG, 0, "Login", "Starting factory Writer");
  125. #elif WORLD
  126. LogWrite(WORLD__DEBUG, 0, "World", "Starting factory Reader");
  127. LogWrite(WORLD__DEBUG, 0, "World", "Starting factory Writer");
  128. #endif
  129. #ifdef WIN32
  130. _beginthread(EQStreamFactoryReaderLoop,0, this);
  131. _beginthread(EQStreamFactoryWriterLoop,0, this);
  132. _beginthread(EQStreamFactoryCombinePacketLoop,0, this);
  133. #else
  134. pthread_create(&t1,NULL,EQStreamFactoryReaderLoop,this);
  135. pthread_create(&t2,NULL,EQStreamFactoryWriterLoop,this);
  136. pthread_create(&t3,NULL,EQStreamFactoryCombinePacketLoop,this);
  137. pthread_detach(t1);
  138. pthread_detach(t2);
  139. pthread_detach(t3);
  140. #endif
  141. return true;
  142. }
  143. EQStream *EQStreamFactory::Pop()
  144. {
  145. if (!NewStreams.size())
  146. return NULL;
  147. EQStream *s=NULL;
  148. //cout << "Pop():Locking MNewStreams" << endl;
  149. MNewStreams.lock();
  150. if (NewStreams.size()) {
  151. s=NewStreams.front();
  152. NewStreams.pop();
  153. s->PutInUse();
  154. }
  155. MNewStreams.unlock();
  156. //cout << "Pop(): Unlocking MNewStreams" << endl;
  157. return s;
  158. }
  159. void EQStreamFactory::Push(EQStream *s)
  160. {
  161. //cout << "Push():Locking MNewStreams" << endl;
  162. MNewStreams.lock();
  163. NewStreams.push(s);
  164. MNewStreams.unlock();
  165. //cout << "Push(): Unlocking MNewStreams" << endl;
  166. }
  167. void EQStreamFactory::ReaderLoop()
  168. {
  169. fd_set readset;
  170. map<string,EQStream *>::iterator stream_itr;
  171. int num;
  172. int length;
  173. unsigned char buffer[2048];
  174. sockaddr_in from;
  175. int socklen=sizeof(sockaddr_in);
  176. timeval sleep_time;
  177. ReaderRunning=true;
  178. while(sock!=-1) {
  179. MReaderRunning.lock();
  180. if (!ReaderRunning)
  181. break;
  182. MReaderRunning.unlock();
  183. FD_ZERO(&readset);
  184. FD_SET(sock,&readset);
  185. sleep_time.tv_sec=30;
  186. sleep_time.tv_usec=0;
  187. if ((num=select(sock+1,&readset,NULL,NULL,&sleep_time))<0) {
  188. // What do we wanna do?
  189. } else if (num==0)
  190. continue;
  191. if (FD_ISSET(sock,&readset)) {
  192. #ifdef WIN32
  193. if ((length=recvfrom(sock,(char*)buffer,sizeof(buffer),0,(struct sockaddr*)&from,(int *)&socklen))<0)
  194. #else
  195. if ((length=recvfrom(sock,buffer,2048,0,(struct sockaddr *)&from,(socklen_t *)&socklen))<0)
  196. #endif
  197. {
  198. // What do we wanna do?
  199. } else {
  200. char temp[25];
  201. sprintf(temp,"%u.%d",ntohl(from.sin_addr.s_addr),ntohs(from.sin_port));
  202. MStreams.lock();
  203. if ((stream_itr=Streams.find(temp))==Streams.end() || buffer[1]==OP_SessionRequest) {
  204. MStreams.unlock();
  205. if (buffer[1]==OP_SessionRequest) {
  206. if(stream_itr != Streams.end() && stream_itr->second)
  207. stream_itr->second->SetState(CLOSED);
  208. EQStream *s=new EQStream(from);
  209. s->SetFactory(this);
  210. s->SetStreamType(StreamType);
  211. Streams[temp]=s;
  212. WriterWork.Signal();
  213. Push(s);
  214. s->Process(buffer,length);
  215. s->SetLastPacketTime(Timer::GetCurrentTime2());
  216. }
  217. } else {
  218. EQStream *curstream = stream_itr->second;
  219. //dont bother processing incoming packets for closed connections
  220. if(curstream->CheckClosed())
  221. curstream = NULL;
  222. else
  223. curstream->PutInUse();
  224. MStreams.unlock();
  225. if(curstream) {
  226. curstream->Process(buffer,length);
  227. curstream->SetLastPacketTime(Timer::GetCurrentTime2());
  228. curstream->ReleaseFromUse();
  229. }
  230. }
  231. }
  232. }
  233. }
  234. }
  235. void EQStreamFactory::CheckTimeout(bool remove_all)
  236. {
  237. //lock streams the entire time were checking timeouts, it should be fast.
  238. MStreams.lock();
  239. unsigned long now=Timer::GetCurrentTime2();
  240. map<string,EQStream *>::iterator stream_itr;
  241. for(stream_itr=Streams.begin();stream_itr!=Streams.end();) {
  242. EQStream *s = stream_itr->second;
  243. EQStreamState state = s->GetState();
  244. if (state==CLOSING && !s->HasOutgoingData()) {
  245. stream_itr->second->SetState(CLOSED);
  246. state = CLOSED;
  247. } else if (s->CheckTimeout(now, STREAM_TIMEOUT)) {
  248. const char* stateString;
  249. switch (state){
  250. case ESTABLISHED:
  251. stateString = "Established";
  252. break;
  253. case CLOSING:
  254. stateString = "Closing";
  255. break;
  256. case CLOSED:
  257. stateString = "Closed";
  258. break;
  259. default:
  260. stateString = "Unknown";
  261. break;
  262. }
  263. LogWrite(WORLD__DEBUG, 0, "World", "Timeout up!, state=%s", stateString);
  264. if (state==ESTABLISHED) {
  265. s->Close();
  266. } else if (state == CLOSING) {
  267. //if we time out in the closing state, just give up
  268. s->SetState(CLOSED);
  269. state = CLOSED;
  270. }
  271. }
  272. //not part of the else so we check it right away on state change
  273. if (remove_all || state==CLOSED) {
  274. if (!remove_all && s->getTimeoutDelays()<2) {
  275. s->addTimeoutDelay();
  276. //give it a little time for everybody to finish with it
  277. } else {
  278. //everybody is done, we can delete it now
  279. #ifdef LOGIN
  280. LogWrite(LOGIN__DEBUG, 0, "Login", "Removing connection...");
  281. #else
  282. LogWrite(WORLD__DEBUG, 0, "World", "Removing connection...");
  283. #endif
  284. map<string,EQStream *>::iterator temp=stream_itr;
  285. stream_itr++;
  286. //let whoever has the stream outside delete it
  287. #ifdef WORLD
  288. client_list.RemoveConnection(temp->second);
  289. #endif
  290. delete temp->second;
  291. Streams.erase(temp);
  292. continue;
  293. }
  294. }
  295. stream_itr++;
  296. }
  297. MStreams.unlock();
  298. }
  299. void EQStreamFactory::CombinePacketLoop(){
  300. deque<EQStream*> combine_que;
  301. CombinePacketRunning = true;
  302. bool packets_waiting = false;
  303. while(sock!=-1) {
  304. if (!CombinePacketRunning)
  305. break;
  306. MStreams.lock();
  307. map<string,EQStream *>::iterator stream_itr;
  308. for(stream_itr=Streams.begin();stream_itr!=Streams.end();stream_itr++) {
  309. if(!stream_itr->second){
  310. continue;
  311. }
  312. if(stream_itr->second->combine_timer && stream_itr->second->combine_timer->Check())
  313. combine_que.push_back(stream_itr->second);
  314. }
  315. EQStream* stream = 0;
  316. packets_waiting = false;
  317. while(combine_que.size()){
  318. stream = combine_que.front();
  319. if(stream->CheckActive()){
  320. if(!stream->CheckCombineQueue())
  321. packets_waiting = true;
  322. }
  323. combine_que.pop_front();
  324. }
  325. MStreams.unlock();
  326. if(!packets_waiting)
  327. Sleep(25);
  328. Sleep(1);
  329. }
  330. }
  331. void EQStreamFactory::WriterLoop()
  332. {
  333. map<string,EQStream *>::iterator stream_itr;
  334. vector<EQStream *> wants_write;
  335. vector<EQStream *>::iterator cur,end;
  336. deque<EQStream*> resend_que;
  337. bool decay=false;
  338. uint32 stream_count;
  339. Timer DecayTimer(20);
  340. WriterRunning=true;
  341. DecayTimer.Enable();
  342. while(sock!=-1) {
  343. Timer::SetCurrentTime();
  344. //if (!havework) {
  345. //WriterWork.Wait();
  346. //}
  347. MWriterRunning.lock();
  348. if (!WriterRunning)
  349. break;
  350. MWriterRunning.unlock();
  351. wants_write.clear();
  352. decay=DecayTimer.Check();
  353. //copy streams into a seperate list so we dont have to keep
  354. //MStreams locked while we are writting
  355. MStreams.lock();
  356. for(stream_itr=Streams.begin();stream_itr!=Streams.end();stream_itr++) {
  357. // If it's time to decay the bytes sent, then let's do it before we try to write
  358. if(!stream_itr->second){
  359. Streams.erase(stream_itr);
  360. break;
  361. }
  362. if (decay)
  363. stream_itr->second->Decay();
  364. if (stream_itr->second->HasOutgoingData()) {
  365. stream_itr->second->PutInUse();
  366. wants_write.push_back(stream_itr->second);
  367. }
  368. if(stream_itr->second->resend_que_timer->Check())
  369. resend_que.push_back(stream_itr->second);
  370. }
  371. MStreams.unlock();
  372. //do the actual writes
  373. cur = wants_write.begin();
  374. end = wants_write.end();
  375. for(; cur != end; cur++) {
  376. (*cur)->Write(sock);
  377. (*cur)->ReleaseFromUse();
  378. }
  379. while(resend_que.size()){
  380. resend_que.front()->CheckResend(sock);
  381. resend_que.pop_front();
  382. }
  383. Sleep(10);
  384. MStreams.lock();
  385. stream_count=Streams.size();
  386. MStreams.unlock();
  387. if (!stream_count) {
  388. //cout << "No streams, waiting on condition" << endl;
  389. WriterWork.Wait();
  390. //cout << "Awake from condition, must have a stream now" << endl;
  391. }
  392. }
  393. }