1 /*
2 * Copyright (C) 2021 Huawei Device Co., Ltd.
3 * Licensed under the Apache License, Version 2.0 (the "License");
4 * you may not use this file except in compliance with the License.
5 * You may obtain a copy of the License at
6 *
7 * http://www.apache.org/licenses/LICENSE-2.0
8 *
9 * Unless required by applicable law or agreed to in writing, software
10 * distributed under the License is distributed on an "AS IS" BASIS,
11 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 * See the License for the specific language governing permissions and
13 * limitations under the License.
14 */
15 #include "session.h"
16 #ifndef TEST_HASH
17 #include "hdc_hash_gen.h"
18 #endif
19 #include "serial_struct.h"
20
21 namespace Hdc {
HdcSessionBase(bool serverOrDaemonIn,size_t uvThreadSize)22 HdcSessionBase::HdcSessionBase(bool serverOrDaemonIn, size_t uvThreadSize)
23 {
24 // print version pid
25 WRITE_LOG(LOG_INFO, "Program running. %s Pid:%u", Base::GetVersion().c_str(), getpid());
26 // server/daemon common initialization code
27 if (uvThreadSize < SIZE_THREAD_POOL_MIN) {
28 uvThreadSize = SIZE_THREAD_POOL_MIN;
29 } else if (uvThreadSize > SIZE_THREAD_POOL_MAX) {
30 uvThreadSize = SIZE_THREAD_POOL_MAX;
31 }
32 threadPoolCount = uvThreadSize;
33 WRITE_LOG(LOG_INFO, "set UV_THREADPOOL_SIZE:%zu", threadPoolCount);
34 string uvThreadEnv("UV_THREADPOOL_SIZE");
35 string uvThreadVal = std::to_string(threadPoolCount);
36 #ifdef _WIN32
37 uvThreadEnv += "=";
38 uvThreadEnv += uvThreadVal;
39 _putenv(uvThreadEnv.c_str());
40 #else
41 setenv(uvThreadEnv.c_str(), uvThreadVal.c_str(), 1);
42 #endif
43 uv_loop_init(&loopMain);
44 WRITE_LOG(LOG_DEBUG, "loopMain init");
45 uv_rwlock_init(&mainAsync);
46 uv_async_init(&loopMain, &asyncMainLoop, MainAsyncCallback);
47 uv_rwlock_init(&lockMapSession);
48 serverOrDaemon = serverOrDaemonIn;
49 ctxUSB = nullptr;
50 wantRestart = false;
51 threadSessionMain = uv_thread_self();
52
53 #ifdef HDC_HOST
54 if (serverOrDaemon) {
55 if (libusb_init((libusb_context **)&ctxUSB) != 0) {
56 ctxUSB = nullptr;
57 WRITE_LOG(LOG_FATAL, "libusb_init failed ctxUSB is nullptr");
58 }
59 }
60 #endif
61 }
62
~HdcSessionBase()63 HdcSessionBase::~HdcSessionBase()
64 {
65 Base::TryCloseHandle((uv_handle_t *)&asyncMainLoop);
66 uv_loop_close(&loopMain);
67 // clear base
68 uv_rwlock_destroy(&mainAsync);
69 uv_rwlock_destroy(&lockMapSession);
70 #ifdef HDC_HOST
71 if (serverOrDaemon and ctxUSB != nullptr) {
72 libusb_exit((libusb_context *)ctxUSB);
73 }
74 #endif
75 WRITE_LOG(LOG_DEBUG, "~HdcSessionBase free sessionRef:%u instance:%s", uint32_t(sessionRef),
76 serverOrDaemon ? "server" : "daemon");
77 }
78
79 // remove step2
TryRemoveTask(HTaskInfo hTask)80 bool HdcSessionBase::TryRemoveTask(HTaskInfo hTask)
81 {
82 if (hTask->taskFree) {
83 WRITE_LOG(LOG_WARN, "TryRemoveTask channelId:%u", hTask->channelId);
84 return true;
85 }
86 bool ret = RemoveInstanceTask(OP_REMOVE, hTask);
87 if (ret) {
88 hTask->taskFree = true;
89 } else {
90 // This is used to check that the memory cannot be cleaned up. If the memory cannot be released, break point
91 // here to see which task has not been released
92 // print task clear
93 }
94 return ret;
95 }
96
97 // remove step1
BeginRemoveTask(HTaskInfo hTask)98 void HdcSessionBase::BeginRemoveTask(HTaskInfo hTask)
99 {
100 StartTraceScope("HdcSessionBase::BeginRemoveTask");
101 if (hTask->taskStop || hTask->taskFree) {
102 WRITE_LOG(LOG_WARN, "BeginRemoveTask channelId:%u taskStop:%d taskFree:%d",
103 hTask->channelId, hTask->taskStop, hTask->taskFree);
104 return;
105 }
106
107 WRITE_LOG(LOG_DEBUG, "BeginRemoveTask taskType:%d channelId:%u", hTask->taskType, hTask->channelId);
108 bool ret = RemoveInstanceTask(OP_CLEAR, hTask);
109 if (!ret) {
110 WRITE_LOG(LOG_INFO, "RemoveInstanceTask false taskType:%d channelId:%u", hTask->taskType, hTask->channelId);
111 }
112 auto taskClassDeleteRetry = [](uv_timer_t *handle) -> void {
113 StartTraceScope("HdcSessionBase::BeginRemoveTask taskClassDeleteRetry");
114 HTaskInfo hTask = (HTaskInfo)handle->data;
115 HdcSessionBase *thisClass = (HdcSessionBase *)hTask->ownerSessionClass;
116 constexpr uint32_t count = 1000;
117 if (hTask->closeRetryCount == 0 || hTask->closeRetryCount > count) {
118 WRITE_LOG(LOG_DEBUG, "TaskDelay task remove retry count %d/%d, taskType:%d channelId:%u, sessionId:%u",
119 hTask->closeRetryCount, GLOBAL_TIMEOUT, hTask->taskType, hTask->channelId, hTask->sessionId);
120 hTask->closeRetryCount = 1;
121 }
122 hTask->closeRetryCount++;
123 if (!thisClass->TryRemoveTask(hTask)) {
124 WRITE_LOG(LOG_WARN, "TaskDelay TryRemoveTask false channelId:%u", hTask->channelId);
125 return;
126 }
127 WRITE_LOG(LOG_DEBUG, "TaskDelay task remove finish, channelId:%u", hTask->channelId);
128 if (hTask != nullptr) {
129 delete hTask;
130 hTask = nullptr;
131 }
132 Base::TryCloseHandle((uv_handle_t *)handle, Base::CloseTimerCallback);
133 };
134 Base::TimerUvTask(hTask->runLoop, hTask, taskClassDeleteRetry, (GLOBAL_TIMEOUT * TIME_BASE) / UV_DEFAULT_INTERVAL);
135
136 hTask->taskStop = true;
137 }
138
139 // Clear all Task or a single Task, the regular situation is stopped first, and the specific class memory is cleaned up
140 // after the end of the LOOP.
141 // When ChannelIdinput == 0, at this time, all of the LOOP ends, all runs in the class end, so directly skip STOP,
142 // physical memory deletion class trimming
ClearOwnTasks(HSession hSession,const uint32_t channelIDInput)143 void HdcSessionBase::ClearOwnTasks(HSession hSession, const uint32_t channelIDInput)
144 {
145 // First case: normal task cleanup process (STOP Remove)
146 // Second: The task is cleaned up, the session ends
147 // Third: The task is cleaned up, and the session is directly over the session.
148 StartTraceScope("HdcSessionBase::ClearOwnTasks");
149 hSession->mapTaskMutex.lock();
150 map<uint32_t, HTaskInfo>::iterator iter;
151 for (iter = hSession->mapTask->begin(); iter != hSession->mapTask->end();) {
152 uint32_t channelId = iter->first;
153 HTaskInfo hTask = iter->second;
154 if (channelIDInput != 0) { // single
155 if (channelIDInput != channelId) {
156 ++iter;
157 continue;
158 }
159 BeginRemoveTask(hTask);
160 WRITE_LOG(LOG_DEBUG, "ClearOwnTasks OP_CLEAR finish, sessionId:%u channelIDInput:%u",
161 hSession->sessionId, channelIDInput);
162 iter = hSession->mapTask->erase(iter);
163 break;
164 }
165 // multi
166 BeginRemoveTask(hTask);
167 iter = hSession->mapTask->erase(iter);
168 }
169 hSession->mapTaskMutex.unlock();
170 }
171
ClearSessions()172 void HdcSessionBase::ClearSessions()
173 {
174 // no need to lock mapSession
175 // broadcast free signal
176 for (auto v : mapSession) {
177 HSession hSession = (HSession)v.second;
178 if (!hSession->isDead) {
179 FreeSession(hSession->sessionId);
180 }
181 }
182 }
183
ReMainLoopForInstanceClear()184 void HdcSessionBase::ReMainLoopForInstanceClear()
185 { // reloop
186 StartTraceScope("HdcSessionBase::ReMainLoopForInstanceClear");
187 auto clearSessionsForFinish = [](uv_idle_t *handle) -> void {
188 HdcSessionBase *thisClass = (HdcSessionBase *)handle->data;
189 if (thisClass->sessionRef > 0) {
190 return;
191 }
192 // all task has been free
193 uv_close((uv_handle_t *)handle, Base::CloseIdleCallback);
194 uv_stop(&thisClass->loopMain);
195 };
196 Base::IdleUvTask(&loopMain, this, clearSessionsForFinish);
197 uv_run(&loopMain, UV_RUN_DEFAULT);
198 };
199
200 #ifdef HDC_SUPPORT_UART
EnumUARTDeviceRegister(UartKickoutZombie kickOut)201 void HdcSessionBase::EnumUARTDeviceRegister(UartKickoutZombie kickOut)
202 {
203 uv_rwlock_rdlock(&lockMapSession);
204 map<uint32_t, HSession>::iterator i;
205 for (i = mapSession.begin(); i != mapSession.end(); ++i) {
206 HSession hs = i->second;
207 if ((hs->connType != CONN_SERIAL) or (hs->hUART == nullptr)) {
208 continue;
209 }
210 kickOut(hs);
211 break;
212 }
213 uv_rwlock_rdunlock(&lockMapSession);
214 }
215 #endif
216
EnumUSBDeviceRegister(void (* pCallBack)(HSession hSession))217 void HdcSessionBase::EnumUSBDeviceRegister(void (*pCallBack)(HSession hSession))
218 {
219 if (!pCallBack) {
220 return;
221 }
222 uv_rwlock_rdlock(&lockMapSession);
223 map<uint32_t, HSession>::iterator i;
224 for (i = mapSession.begin(); i != mapSession.end(); ++i) {
225 HSession hs = i->second;
226 if (hs->connType != CONN_USB) {
227 continue;
228 }
229 if (hs->hUSB == nullptr) {
230 continue;
231 }
232 if (pCallBack) {
233 pCallBack(hs);
234 }
235 break;
236 }
237 uv_rwlock_rdunlock(&lockMapSession);
238 }
239
240 // The PC side gives the device information, determines if the USB device is registered
241 // PDEV and Busid Devid two choices
QueryUSBDeviceRegister(void * pDev,uint8_t busIDIn,uint8_t devIDIn)242 HSession HdcSessionBase::QueryUSBDeviceRegister(void *pDev, uint8_t busIDIn, uint8_t devIDIn)
243 {
244 #ifdef HDC_HOST
245 libusb_device *dev = (libusb_device *)pDev;
246 HSession hResult = nullptr;
247 if (!mapSession.size()) {
248 return nullptr;
249 }
250 uint8_t busId = 0;
251 uint8_t devId = 0;
252 if (pDev) {
253 busId = libusb_get_bus_number(dev);
254 devId = libusb_get_device_address(dev);
255 } else {
256 busId = busIDIn;
257 devId = devIDIn;
258 }
259 uv_rwlock_rdlock(&lockMapSession);
260 map<uint32_t, HSession>::iterator i;
261 for (i = mapSession.begin(); i != mapSession.end(); ++i) {
262 HSession hs = i->second;
263 if (hs->connType == CONN_USB) {
264 continue;
265 }
266 if (hs->hUSB == nullptr) {
267 continue;
268 }
269 if (hs->hUSB->devId != devId || hs->hUSB->busId != busId) {
270 continue;
271 }
272 hResult = hs;
273 break;
274 }
275 uv_rwlock_rdunlock(&lockMapSession);
276 return hResult;
277 #else
278 return nullptr;
279 #endif
280 }
281
AsyncMainLoopTask(uv_idle_t * handle)282 void HdcSessionBase::AsyncMainLoopTask(uv_idle_t *handle)
283 {
284 AsyncParam *param = (AsyncParam *)handle->data;
285 HdcSessionBase *thisClass = (HdcSessionBase *)param->thisClass;
286 switch (param->method) {
287 case ASYNC_FREE_SESSION:
288 // Destruction is unified in the main thread
289 thisClass->FreeSession(param->sid);
290 break;
291 case ASYNC_STOP_MAINLOOP:
292 uv_stop(&thisClass->loopMain);
293 break;
294 default:
295 break;
296 }
297 if (param->data) {
298 delete[]((uint8_t *)param->data);
299 }
300 delete param;
301 param = nullptr;
302 Base::TryCloseHandle((uv_handle_t *)handle, Base::CloseIdleCallback);
303 }
304
MainAsyncCallback(uv_async_t * handle)305 void HdcSessionBase::MainAsyncCallback(uv_async_t *handle)
306 {
307 HdcSessionBase *thisClass = (HdcSessionBase *)handle->data;
308 list<void *>::iterator i;
309 list<void *> &lst = thisClass->lstMainThreadOP;
310 uv_rwlock_wrlock(&thisClass->mainAsync);
311 for (i = lst.begin(); i != lst.end();) {
312 AsyncParam *param = (AsyncParam *)*i;
313 Base::IdleUvTask(&thisClass->loopMain, param, AsyncMainLoopTask);
314 i = lst.erase(i);
315 }
316 uv_rwlock_wrunlock(&thisClass->mainAsync);
317 }
318
PushAsyncMessage(const uint32_t sessionId,const uint8_t method,const void * data,const int dataSize)319 void HdcSessionBase::PushAsyncMessage(const uint32_t sessionId, const uint8_t method, const void *data,
320 const int dataSize)
321 {
322 AsyncParam *param = new AsyncParam();
323 if (!param) {
324 return;
325 }
326 param->sid = sessionId;
327 param->thisClass = this;
328 param->method = method;
329 if (dataSize > 0) {
330 param->dataSize = dataSize;
331 param->data = new uint8_t[param->dataSize]();
332 if (!param->data) {
333 delete param;
334 return;
335 }
336 if (memcpy_s((uint8_t *)param->data, param->dataSize, data, dataSize)) {
337 delete[]((uint8_t *)param->data);
338 delete param;
339 return;
340 }
341 }
342
343 asyncMainLoop.data = this;
344 uv_rwlock_wrlock(&mainAsync);
345 lstMainThreadOP.push_back(param);
346 uv_rwlock_wrunlock(&mainAsync);
347 uv_async_send(&asyncMainLoop);
348 }
349
WorkerPendding()350 void HdcSessionBase::WorkerPendding()
351 {
352 uv_run(&loopMain, UV_RUN_DEFAULT);
353 ClearInstanceResource();
354 }
355
MallocSessionByConnectType(HSession hSession)356 int HdcSessionBase::MallocSessionByConnectType(HSession hSession)
357 {
358 int ret = 0;
359 switch (hSession->connType) {
360 case CONN_TCP: {
361 uv_tcp_init(&loopMain, &hSession->hWorkTCP);
362 ++hSession->uvHandleRef;
363 hSession->hWorkTCP.data = hSession;
364 break;
365 }
366 case CONN_USB: {
367 // Some members need to be placed at the primary thread
368 HUSB hUSB = new HdcUSB();
369 if (!hUSB) {
370 ret = -1;
371 break;
372 }
373 hSession->hUSB = hUSB;
374 hSession->hUSB->wMaxPacketSizeSend = MAX_PACKET_SIZE_HISPEED;
375 break;
376 }
377 #ifdef HDC_SUPPORT_UART
378 case CONN_SERIAL: {
379 HUART hUART = new HdcUART();
380 if (!hUART) {
381 ret = -1;
382 break;
383 }
384 hSession->hUART = hUART;
385 break;
386 }
387 #endif // HDC_SUPPORT_UART
388 default:
389 ret = -1;
390 break;
391 }
392 return ret;
393 }
394
395 // Avoid unit test when client\server\daemon on the same host, maybe get the same ID value
GetSessionPseudoUid()396 uint32_t HdcSessionBase::GetSessionPseudoUid()
397 {
398 uint32_t uid = 0;
399 do {
400 uid = static_cast<uint32_t>(Base::GetRandom());
401 } while (AdminSession(OP_QUERY, uid, nullptr) != nullptr);
402 return uid;
403 }
404
405 // when client 0 to automatic generated，when daemon First place 1 followed by
MallocSession(bool serverOrDaemon,const ConnType connType,void * classModule,uint32_t sessionId)406 HSession HdcSessionBase::MallocSession(bool serverOrDaemon, const ConnType connType, void *classModule,
407 uint32_t sessionId)
408 {
409 #ifdef CONFIG_USE_JEMALLOC_DFX_INIF
410 mallopt(M_DELAYED_FREE, M_DELAYED_FREE_DISABLE);
411 mallopt(M_SET_THREAD_CACHE, M_THREAD_CACHE_DISABLE);
412 #endif
413 HSession hSession = new(std::nothrow) HdcSession();
414 if (!hSession) {
415 WRITE_LOG(LOG_FATAL, "MallocSession new hSession failed");
416 return nullptr;
417 }
418 int ret = 0;
419 ++sessionRef;
420 hSession->classInstance = this;
421 hSession->connType = connType;
422 hSession->classModule = classModule;
423 hSession->isDead = false;
424 hSession->sessionId = ((sessionId == 0) ? GetSessionPseudoUid() : sessionId);
425 hSession->serverOrDaemon = serverOrDaemon;
426 hSession->hWorkThread = uv_thread_self();
427 hSession->mapTask = new(std::nothrow) map<uint32_t, HTaskInfo>();
428 if (hSession->mapTask == nullptr) {
429 WRITE_LOG(LOG_FATAL, "MallocSession new hSession->mapTask failed");
430 delete hSession;
431 hSession = nullptr;
432 return nullptr;
433 }
434 hSession->listKey = new(std::nothrow) list<void *>;
435 if (hSession->listKey == nullptr) {
436 WRITE_LOG(LOG_FATAL, "MallocSession new hSession->listKey failed");
437 delete hSession;
438 hSession = nullptr;
439 return nullptr;
440 }
441 uv_loop_init(&hSession->childLoop);
442 hSession->uvHandleRef = 0;
443 // pullup child
444 WRITE_LOG(LOG_DEBUG, "HdcSessionBase NewSession, sessionId:%u, connType:%d.",
445 hSession->sessionId, hSession->connType);
446 ++hSession->uvHandleRef;
447 Base::CreateSocketPair(hSession->ctrlFd);
448 size_t handleSize = sizeof(uv_poll_t);
449 hSession->pollHandle[STREAM_WORK] = (uv_poll_t *)malloc(handleSize);
450 hSession->pollHandle[STREAM_MAIN] = (uv_poll_t *)malloc(handleSize);
451 uv_poll_t *pollHandleMain = hSession->pollHandle[STREAM_MAIN];
452 if (pollHandleMain == nullptr || hSession->pollHandle[STREAM_WORK] == nullptr) {
453 WRITE_LOG(LOG_FATAL, "MallocSession malloc hSession->pollHandle failed");
454 delete hSession;
455 hSession = nullptr;
456 return nullptr;
457 }
458 uv_poll_init_socket(&loopMain, pollHandleMain, hSession->ctrlFd[STREAM_MAIN]);
459 uv_poll_start(pollHandleMain, UV_READABLE, ReadCtrlFromSession);
460 hSession->pollHandle[STREAM_MAIN]->data = hSession;
461 hSession->pollHandle[STREAM_WORK]->data = hSession;
462 // Activate USB DAEMON's data channel, may not for use
463 uv_tcp_init(&loopMain, &hSession->dataPipe[STREAM_MAIN]);
464 (void)memset_s(&hSession->dataPipe[STREAM_WORK], sizeof(hSession->dataPipe[STREAM_WORK]),
465 0, sizeof(uv_tcp_t));
466 ++hSession->uvHandleRef;
467 Base::CreateSocketPair(hSession->dataFd);
468 uv_tcp_open(&hSession->dataPipe[STREAM_MAIN], hSession->dataFd[STREAM_MAIN]);
469 hSession->dataPipe[STREAM_MAIN].data = hSession;
470 hSession->dataPipe[STREAM_WORK].data = hSession;
471 #ifdef HDC_HOST
472 Base::SetTcpOptions(&hSession->dataPipe[STREAM_MAIN], HOST_SOCKETPAIR_SIZE);
473 #else
474 Base::SetTcpOptions(&hSession->dataPipe[STREAM_MAIN]);
475 #endif
476 ret = MallocSessionByConnectType(hSession);
477 if (ret) {
478 delete hSession;
479 hSession = nullptr;
480 } else {
481 AdminSession(OP_ADD, hSession->sessionId, hSession);
482 }
483 return hSession;
484 }
485
FreeSessionByConnectType(HSession hSession)486 void HdcSessionBase::FreeSessionByConnectType(HSession hSession)
487 {
488 WRITE_LOG(LOG_DEBUG, "FreeSessionByConnectType %s", hSession->ToDebugString().c_str());
489
490 if (hSession->connType == CONN_USB) {
491 // ibusb All context is applied for sub-threaded, so it needs to be destroyed in the subline
492 if (!hSession->hUSB) {
493 return;
494 }
495 HUSB hUSB = hSession->hUSB;
496 if (!hUSB) {
497 return;
498 }
499 #ifdef HDC_HOST
500 if (hUSB->devHandle) {
501 libusb_release_interface(hUSB->devHandle, hUSB->interfaceNumber);
502 libusb_close(hUSB->devHandle);
503 hUSB->devHandle = nullptr;
504 }
505 #else
506 Base::CloseFd(hUSB->bulkIn);
507 Base::CloseFd(hUSB->bulkOut);
508 #endif
509 delete hSession->hUSB;
510 hSession->hUSB = nullptr;
511 }
512 #ifdef HDC_SUPPORT_UART
513 if (CONN_SERIAL == hSession->connType) {
514 if (!hSession->hUART) {
515 return;
516 }
517 HUART hUART = hSession->hUART;
518 if (!hUART) {
519 return;
520 }
521 HdcUARTBase *uartBase = (HdcUARTBase *)hSession->classModule;
522 // tell uart session will be free
523 uartBase->StopSession(hSession);
524 #ifdef HDC_HOST
525 #ifdef HOST_MINGW
526 if (hUART->devUartHandle != INVALID_HANDLE_VALUE) {
527 CloseHandle(hUART->devUartHandle);
528 hUART->devUartHandle = INVALID_HANDLE_VALUE;
529 }
530 #elif defined(HOST_LINUX)
531 Base::CloseFd(hUART->devUartHandle);
532 #endif // _WIN32
533 #endif
534 delete hSession->hUART;
535 hSession->hUART = nullptr;
536 }
537 #endif
538 }
539
540 // work when libuv-handle at struct of HdcSession has all callback finished
FreeSessionFinally(uv_idle_t * handle)541 void HdcSessionBase::FreeSessionFinally(uv_idle_t *handle)
542 {
543 HSession hSession = (HSession)handle->data;
544 HdcSessionBase *thisClass = (HdcSessionBase *)hSession->classInstance;
545 if (hSession->uvHandleRef > 0) {
546 WRITE_LOG(LOG_DEBUG, "FreeSessionFinally uvHandleRef:%d sessionId:%u",
547 hSession->uvHandleRef, hSession->sessionId);
548 return;
549 }
550 // Notify Server or Daemon, just UI or display commandline
551 thisClass->NotifyInstanceSessionFree(hSession, true);
552 // all hsession uv handle has been clear
553 thisClass->AdminSession(OP_REMOVE, hSession->sessionId, nullptr);
554 WRITE_LOG(LOG_DEBUG, "!!!FreeSessionFinally sessionId:%u finish", hSession->sessionId);
555 HdcAuth::FreeKey(!hSession->serverOrDaemon, hSession->listKey);
556 delete hSession;
557 hSession = nullptr; // fix CodeMars SetNullAfterFree issue
558 Base::TryCloseHandle((const uv_handle_t *)handle, Base::CloseIdleCallback);
559 --thisClass->sessionRef;
560 }
561
562 // work when child-work thread finish
FreeSessionContinue(HSession hSession)563 void HdcSessionBase::FreeSessionContinue(HSession hSession)
564 {
565 auto closeSessionTCPHandle = [](uv_handle_t *handle) -> void {
566 HSession hSession = (HSession)handle->data;
567 --hSession->uvHandleRef;
568 Base::TryCloseHandle((uv_handle_t *)handle);
569 if (handle == reinterpret_cast<uv_handle_t *>(hSession->pollHandle[STREAM_MAIN])) {
570 Base::CloseFd(hSession->ctrlFd[STREAM_MAIN]);
571 Base::CloseFd(hSession->ctrlFd[STREAM_WORK]);
572 free(hSession->pollHandle[STREAM_MAIN]);
573 }
574 };
575 if (hSession->connType == CONN_TCP) {
576 // Turn off TCP to prevent continuing writing
577 Base::TryCloseHandle((uv_handle_t *)&hSession->hWorkTCP, true, closeSessionTCPHandle);
578 }
579 hSession->availTailIndex = 0;
580 if (hSession->ioBuf) {
581 delete[] hSession->ioBuf;
582 hSession->ioBuf = nullptr;
583 }
584 Base::TryCloseHandle((uv_handle_t *)hSession->pollHandle[STREAM_MAIN], true, closeSessionTCPHandle);
585 Base::TryCloseHandle((uv_handle_t *)&hSession->dataPipe[STREAM_MAIN], true, closeSessionTCPHandle);
586 FreeSessionByConnectType(hSession);
587 // finish
588 Base::IdleUvTask(&loopMain, hSession, FreeSessionFinally);
589 }
590
FreeSessionOpeate(uv_timer_t * handle)591 void HdcSessionBase::FreeSessionOpeate(uv_timer_t *handle)
592 {
593 StartTraceScope("HdcSessionBase::FreeSessionOpeate");
594 HSession hSession = (HSession)handle->data;
595 HdcSessionBase *thisClass = (HdcSessionBase *)hSession->classInstance;
596 if (hSession->ref > 0) {
597 WRITE_LOG(LOG_WARN, "FreeSessionOpeate ref:%u > 0", uint32_t(hSession->ref));
598 return;
599 }
600 WRITE_LOG(LOG_DEBUG, "FreeSessionOpeate ref:%u", uint32_t(hSession->ref));
601 #ifdef HDC_HOST
602 if (hSession->hUSB != nullptr
603 && (!hSession->hUSB->hostBulkIn.isShutdown || !hSession->hUSB->hostBulkOut.isShutdown)) {
604 HdcUSBBase *pUSB = ((HdcUSBBase *)hSession->classModule);
605 pUSB->CancelUsbIo(hSession);
606 return;
607 }
608 #endif
609 // wait workthread to free
610 if (hSession->pollHandle[STREAM_WORK]->loop) {
611 auto ctrl = BuildCtrlString(SP_STOP_SESSION, 0, nullptr, 0);
612 Base::SendToPollFd(hSession->ctrlFd[STREAM_MAIN], ctrl.data(), ctrl.size());
613 WRITE_LOG(LOG_DEBUG, "FreeSessionOpeate, send workthread for free. sessionId:%u", hSession->sessionId);
614 auto callbackCheckFreeSessionContinue = [](uv_timer_t *handle) -> void {
615 HSession hSession = (HSession)handle->data;
616 HdcSessionBase *thisClass = (HdcSessionBase *)hSession->classInstance;
617 if (!hSession->childCleared) {
618 WRITE_LOG(LOG_DEBUG, "FreeSessionOpeate childCleared:%d sessionId:%u",
619 hSession->childCleared, hSession->sessionId);
620 return;
621 }
622 Base::TryCloseHandle((uv_handle_t *)handle, Base::CloseTimerCallback);
623 thisClass->FreeSessionContinue(hSession);
624 };
625 Base::TimerUvTask(&thisClass->loopMain, hSession, callbackCheckFreeSessionContinue);
626 } else {
627 thisClass->FreeSessionContinue(hSession);
628 }
629 Base::TryCloseHandle((uv_handle_t *)handle, Base::CloseTimerCallback);
630 }
631
FreeSession(const uint32_t sessionId)632 void HdcSessionBase::FreeSession(const uint32_t sessionId)
633 {
634 StartTraceScope("HdcSessionBase::FreeSession");
635 if (threadSessionMain != uv_thread_self()) {
636 PushAsyncMessage(sessionId, ASYNC_FREE_SESSION, nullptr, 0);
637 return;
638 }
639 HSession hSession = AdminSession(OP_QUERY, sessionId, nullptr);
640 WRITE_LOG(LOG_DEBUG, "Begin to free session, sessionid:%u", sessionId);
641 do {
642 if (!hSession || hSession->isDead) {
643 WRITE_LOG(LOG_WARN, "FreeSession hSession nullptr or isDead sessionId:%u", sessionId);
644 break;
645 }
646 hSession->isDead = true;
647 Base::TimerUvTask(&loopMain, hSession, FreeSessionOpeate);
648 NotifyInstanceSessionFree(hSession, false);
649 WRITE_LOG(LOG_DEBUG, "FreeSession sessionId:%u ref:%u", hSession->sessionId, uint32_t(hSession->ref));
650 } while (false);
651 }
652
AdminSession(const uint8_t op,const uint32_t sessionId,HSession hInput)653 HSession HdcSessionBase::AdminSession(const uint8_t op, const uint32_t sessionId, HSession hInput)
654 {
655 HSession hRet = nullptr;
656 switch (op) {
657 case OP_ADD:
658 uv_rwlock_wrlock(&lockMapSession);
659 mapSession[sessionId] = hInput;
660 uv_rwlock_wrunlock(&lockMapSession);
661 break;
662 case OP_REMOVE:
663 uv_rwlock_wrlock(&lockMapSession);
664 mapSession.erase(sessionId);
665 uv_rwlock_wrunlock(&lockMapSession);
666 break;
667 case OP_QUERY:
668 uv_rwlock_rdlock(&lockMapSession);
669 if (mapSession.count(sessionId)) {
670 hRet = mapSession[sessionId];
671 }
672 uv_rwlock_rdunlock(&lockMapSession);
673 break;
674 case OP_QUERY_REF:
675 uv_rwlock_wrlock(&lockMapSession);
676 if (mapSession.count(sessionId)) {
677 hRet = mapSession[sessionId];
678 ++hRet->ref;
679 }
680 uv_rwlock_wrunlock(&lockMapSession);
681 break;
682 case OP_UPDATE:
683 uv_rwlock_wrlock(&lockMapSession);
684 // remove old
685 mapSession.erase(sessionId);
686 mapSession[hInput->sessionId] = hInput;
687 uv_rwlock_wrunlock(&lockMapSession);
688 break;
689 case OP_VOTE_RESET:
690 if (mapSession.count(sessionId) == 0) {
691 break;
692 }
693 bool needReset;
694 if (serverOrDaemon) {
695 uv_rwlock_wrlock(&lockMapSession);
696 hRet = mapSession[sessionId];
697 hRet->voteReset = true;
698 needReset = true;
699 for (auto &kv : mapSession) {
700 if (sessionId == kv.first) {
701 continue;
702 }
703 WRITE_LOG(LOG_DEBUG, "session:%u vote reset, session %u is %s",
704 sessionId, kv.first, kv.second->voteReset ? "YES" : "NO");
705 if (!kv.second->voteReset) {
706 needReset = false;
707 }
708 }
709 uv_rwlock_wrunlock(&lockMapSession);
710 } else {
711 needReset = true;
712 }
713 if (needReset) {
714 WRITE_LOG(LOG_FATAL, "!! session:%u vote reset, passed unanimously !!", sessionId);
715 abort();
716 }
717 break;
718 default:
719 break;
720 }
721 return hRet;
722 }
723
DumpTasksInfo(map<uint32_t,HTaskInfo> & mapTask)724 void HdcSessionBase::DumpTasksInfo(map<uint32_t, HTaskInfo> &mapTask)
725 {
726 int idx = 1;
727 for (auto t : mapTask) {
728 HTaskInfo ti = t.second;
729 WRITE_LOG(LOG_WARN, "%d: channelId: %lu, type: %d, closeRetry: %d\n",
730 idx++, ti->channelId, ti->taskType, ti->closeRetryCount);
731 }
732 }
733
734 // All in the corresponding sub-thread, no need locks
AdminTask(const uint8_t op,HSession hSession,const uint32_t channelId,HTaskInfo hInput)735 HTaskInfo HdcSessionBase::AdminTask(const uint8_t op, HSession hSession, const uint32_t channelId, HTaskInfo hInput)
736 {
737 HTaskInfo hRet = nullptr;
738 map<uint32_t, HTaskInfo> &mapTask = *hSession->mapTask;
739
740 switch (op) {
741 case OP_ADD:
742 hRet = mapTask[channelId];
743 if (hRet != nullptr) {
744 delete hRet;
745 }
746 mapTask[channelId] = hInput;
747 hRet = hInput;
748
749 WRITE_LOG(LOG_DEBUG, "AdminTask add session %u, channelId %u, mapTask size: %zu",
750 hSession->sessionId, channelId, mapTask.size());
751
752 break;
753 case OP_REMOVE:
754 mapTask.erase(channelId);
755 WRITE_LOG(LOG_DEBUG, "AdminTask rm session %u, channelId %u, mapTask size: %zu",
756 hSession->sessionId, channelId, mapTask.size());
757 break;
758 case OP_QUERY:
759 if (mapTask.count(channelId)) {
760 hRet = mapTask[channelId];
761 }
762 break;
763 case OP_VOTE_RESET:
764 AdminSession(op, hSession->sessionId, nullptr);
765 break;
766 default:
767 break;
768 }
769 return hRet;
770 }
771
SendByProtocol(HSession hSession,uint8_t * bufPtr,const int bufLen,bool echo)772 int HdcSessionBase::SendByProtocol(HSession hSession, uint8_t *bufPtr, const int bufLen, bool echo)
773 {
774 StartTraceScope("HdcSessionBase::SendByProtocol");
775 if (hSession->isDead) {
776 delete[] bufPtr;
777 WRITE_LOG(LOG_WARN, "SendByProtocol session dead error");
778 return ERR_SESSION_NOFOUND;
779 }
780 int ret = 0;
781 switch (hSession->connType) {
782 case CONN_TCP: {
783 HdcTCPBase *pTCP = ((HdcTCPBase *)hSession->classModule);
784 if (echo && !hSession->serverOrDaemon) {
785 ret = pTCP->WriteUvTcpFd(&hSession->hChildWorkTCP, bufPtr, bufLen);
786 } else {
787 if (hSession->hWorkThread == uv_thread_self()) {
788 ret = pTCP->WriteUvTcpFd(&hSession->hWorkTCP, bufPtr, bufLen);
789 } else {
790 ret = pTCP->WriteUvTcpFd(&hSession->hChildWorkTCP, bufPtr, bufLen);
791 }
792 }
793 break;
794 }
795 case CONN_USB: {
796 HdcUSBBase *pUSB = ((HdcUSBBase *)hSession->classModule);
797 ret = pUSB->SendUSBBlock(hSession, bufPtr, bufLen);
798 delete[] bufPtr;
799 break;
800 }
801 #ifdef HDC_SUPPORT_UART
802 case CONN_SERIAL: {
803 HdcUARTBase *pUART = ((HdcUARTBase *)hSession->classModule);
804 ret = pUART->SendUARTData(hSession, bufPtr, bufLen);
805 delete[] bufPtr;
806 break;
807 }
808 #endif
809 default:
810 break;
811 }
812 return ret;
813 }
814
Send(const uint32_t sessionId,const uint32_t channelId,const uint16_t commandFlag,const uint8_t * data,const int dataSize)815 int HdcSessionBase::Send(const uint32_t sessionId, const uint32_t channelId, const uint16_t commandFlag,
816 const uint8_t *data, const int dataSize)
817 {
818 StartTraceScope("HdcSessionBase::Send");
819 HSession hSession = AdminSession(OP_QUERY, sessionId, nullptr);
820 if (!hSession) {
821 WRITE_LOG(LOG_DEBUG, "Send to offline device, drop it, sessionId:%u", sessionId);
822 return ERR_SESSION_NOFOUND;
823 }
824 PayloadProtect protectBuf; // noneed convert to big-endian
825 protectBuf.channelId = channelId;
826 protectBuf.commandFlag = commandFlag;
827 protectBuf.checkSum = (ENABLE_IO_CHECKSUM && dataSize > 0) ? Base::CalcCheckSum(data, dataSize) : 0;
828 protectBuf.vCode = payloadProtectStaticVcode;
829 string s = SerialStruct::SerializeToString(protectBuf);
830 // reserve for encrypt here
831 // xx-encrypt
832
833 PayloadHead payloadHead = {}; // need convert to big-endian
834 payloadHead.flag[0] = PACKET_FLAG.at(0);
835 payloadHead.flag[1] = PACKET_FLAG.at(1);
836 payloadHead.protocolVer = VER_PROTOCOL;
837 payloadHead.headSize = htons(s.size());
838 payloadHead.dataSize = htonl(dataSize);
839 int finalBufSize = sizeof(PayloadHead) + s.size() + dataSize;
840 uint8_t *finayBuf = new(std::nothrow) uint8_t[finalBufSize]();
841 if (finayBuf == nullptr) {
842 WRITE_LOG(LOG_WARN, "send allocmem err");
843 return ERR_BUF_ALLOC;
844 }
845 bool bufRet = false;
846 do {
847 if (memcpy_s(finayBuf, sizeof(PayloadHead), reinterpret_cast<uint8_t *>(&payloadHead), sizeof(PayloadHead))) {
848 WRITE_LOG(LOG_WARN, "send copyhead err for dataSize:%d", dataSize);
849 break;
850 }
851 if (memcpy_s(finayBuf + sizeof(PayloadHead), s.size(),
852 reinterpret_cast<uint8_t *>(const_cast<char *>(s.c_str())), s.size())) {
853 WRITE_LOG(LOG_WARN, "send copyProtbuf err for dataSize:%d", dataSize);
854 break;
855 }
856 if (dataSize > 0 && memcpy_s(finayBuf + sizeof(PayloadHead) + s.size(), dataSize, data, dataSize)) {
857 WRITE_LOG(LOG_WARN, "send copyDatabuf err for dataSize:%d", dataSize);
858 break;
859 }
860 bufRet = true;
861 } while (false);
862 if (!bufRet) {
863 delete[] finayBuf;
864 WRITE_LOG(LOG_WARN, "send copywholedata err for dataSize:%d", dataSize);
865 return ERR_BUF_COPY;
866 }
867 if (CMD_KERNEL_ECHO == commandFlag) {
868 return SendByProtocol(hSession, finayBuf, finalBufSize, true);
869 } else {
870 return SendByProtocol(hSession, finayBuf, finalBufSize);
871 }
872 }
873
DecryptPayload(HSession hSession,PayloadHead * payloadHeadBe,uint8_t * encBuf)874 int HdcSessionBase::DecryptPayload(HSession hSession, PayloadHead *payloadHeadBe, uint8_t *encBuf)
875 {
876 StartTraceScope("HdcSessionBase::DecryptPayload");
877 PayloadProtect protectBuf = {};
878 uint16_t headSize = ntohs(payloadHeadBe->headSize);
879 int dataSize = ntohl(payloadHeadBe->dataSize);
880 string encString(reinterpret_cast<char *>(encBuf), headSize);
881 SerialStruct::ParseFromString(protectBuf, encString);
882 if (protectBuf.vCode != payloadProtectStaticVcode) {
883 WRITE_LOG(LOG_FATAL, "Session recv static vcode failed");
884 return ERR_BUF_CHECK;
885 }
886 uint8_t *data = encBuf + headSize;
887 if (ENABLE_IO_CHECKSUM && protectBuf.checkSum != 0 && (protectBuf.checkSum != Base::CalcCheckSum(data, dataSize))) {
888 WRITE_LOG(LOG_FATAL, "Session recv CalcCheckSum failed");
889 return ERR_BUF_CHECK;
890 }
891 if (!FetchCommand(hSession, protectBuf.channelId, protectBuf.commandFlag, data, dataSize)) {
892 WRITE_LOG(LOG_WARN, "FetchCommand failed: channelId %x commandFlag %x",
893 protectBuf.channelId, protectBuf.commandFlag);
894 return ERR_GENERIC;
895 }
896 return RET_SUCCESS;
897 }
898
OnRead(HSession hSession,uint8_t * bufPtr,const int bufLen)899 int HdcSessionBase::OnRead(HSession hSession, uint8_t *bufPtr, const int bufLen)
900 {
901 int ret = ERR_GENERIC;
902 StartTraceScope("HdcSessionBase::OnRead");
903 if (memcmp(bufPtr, PACKET_FLAG.c_str(), PACKET_FLAG.size())) {
904 WRITE_LOG(LOG_FATAL, "PACKET_FLAG incorrect %x %x", bufPtr[0], bufPtr[1]);
905 return ERR_BUF_CHECK;
906 }
907 struct PayloadHead *payloadHead = reinterpret_cast<struct PayloadHead *>(bufPtr);
908 // to prevent integer overflow caused by the add operation of two input num
909 uint64_t payloadHeadSize = static_cast<uint64_t>(ntohl(payloadHead->dataSize)) +
910 static_cast<uint64_t>(ntohs(payloadHead->headSize));
911 int packetHeadSize = sizeof(struct PayloadHead);
912 if (payloadHeadSize == 0 || payloadHeadSize > static_cast<uint64_t>(HDC_BUF_MAX_BYTES)) {
913 WRITE_LOG(LOG_FATAL, "Packet size incorrect");
914 return ERR_BUF_CHECK;
915 }
916
917 // 0 < payloadHeadSize < HDC_BUF_MAX_BYTES
918 int tobeReadLen = static_cast<int>(payloadHeadSize);
919 if (bufLen - packetHeadSize < tobeReadLen) {
920 return 0;
921 }
922 if (DecryptPayload(hSession, payloadHead, bufPtr + packetHeadSize)) {
923 WRITE_LOG(LOG_WARN, "decrypt plhead error");
924 return ERR_BUF_CHECK;
925 }
926 ret = packetHeadSize + tobeReadLen;
927 return ret;
928 }
929
930 // Returns <0 error;> 0 receives the number of bytes; 0 untreated
FetchIOBuf(HSession hSession,uint8_t * ioBuf,int read)931 int HdcSessionBase::FetchIOBuf(HSession hSession, uint8_t *ioBuf, int read)
932 {
933 HdcSessionBase *ptrConnect = (HdcSessionBase *)hSession->classInstance;
934 int indexBuf = 0;
935 int childRet = 0;
936 StartTraceScope("HdcSessionBase::FetchIOBuf");
937 if (read < 0) {
938 constexpr int bufSize = 1024;
939 char buf[bufSize] = { 0 };
940 uv_strerror_r(read, buf, bufSize);
941 WRITE_LOG(LOG_FATAL, "FetchIOBuf read io failed,%s", buf);
942 return ERR_IO_FAIL;
943 }
944 hSession->availTailIndex += read;
945 while (!hSession->isDead && hSession->availTailIndex > static_cast<int>(sizeof(PayloadHead))) {
946 childRet = ptrConnect->OnRead(hSession, ioBuf + indexBuf, hSession->availTailIndex);
947 if (childRet > 0) {
948 hSession->availTailIndex -= childRet;
949 indexBuf += childRet;
950 } else if (childRet == 0) {
951 // Not enough a IO
952 break;
953 } else { // <0
954 WRITE_LOG(LOG_FATAL, "FetchIOBuf error childRet:%d sessionId:%u", childRet, hSession->sessionId);
955 hSession->availTailIndex = 0; // Preventing malicious data packages
956 indexBuf = ERR_BUF_SIZE;
957 break;
958 }
959 // It may be multi-time IO to merge in a BUF, need to loop processing
960 }
961 if (indexBuf > 0 && hSession->availTailIndex > 0) {
962 if (memmove_s(hSession->ioBuf, hSession->bufSize, hSession->ioBuf + indexBuf, hSession->availTailIndex)
963 != EOK) {
964 return ERR_BUF_COPY;
965 };
966 uint8_t *bufToZero = reinterpret_cast<uint8_t *>(hSession->ioBuf + hSession->availTailIndex);
967 Base::ZeroBuf(bufToZero, hSession->bufSize - hSession->availTailIndex);
968 }
969 return indexBuf;
970 }
971
AllocCallback(uv_handle_t * handle,size_t sizeWanted,uv_buf_t * buf)972 void HdcSessionBase::AllocCallback(uv_handle_t *handle, size_t sizeWanted, uv_buf_t *buf)
973 {
974 HSession context = (HSession)handle->data;
975 Base::ReallocBuf(&context->ioBuf, &context->bufSize, HDC_SOCKETPAIR_SIZE);
976 buf->base = (char *)context->ioBuf + context->availTailIndex;
977 int size = context->bufSize - context->availTailIndex;
978 buf->len = std::min(size, static_cast<int>(sizeWanted));
979 }
980
FinishWriteSessionTCP(uv_write_t * req,int status)981 void HdcSessionBase::FinishWriteSessionTCP(uv_write_t *req, int status)
982 {
983 HSession hSession = (HSession)req->handle->data;
984 --hSession->ref;
985 HdcSessionBase *thisClass = (HdcSessionBase *)hSession->classInstance;
986 if (status < 0) {
987 WRITE_LOG(LOG_WARN, "FinishWriteSessionTCP status:%d sessionId:%u isDead:%d ref:%u",
988 status, hSession->sessionId, hSession->isDead, uint32_t(hSession->ref));
989 Base::TryCloseHandle((uv_handle_t *)req->handle);
990 if (!hSession->isDead && !hSession->ref) {
991 WRITE_LOG(LOG_DEBUG, "FinishWriteSessionTCP freesession :%u", hSession->sessionId);
992 thisClass->FreeSession(hSession->sessionId);
993 }
994 }
995 delete[]((uint8_t *)req->data);
996 delete req;
997 }
998
DispatchSessionThreadCommand(HSession hSession,const uint8_t * baseBuf,const int bytesIO)999 bool HdcSessionBase::DispatchSessionThreadCommand(HSession hSession, const uint8_t *baseBuf,
1000 const int bytesIO)
1001 {
1002 bool ret = true;
1003 uint8_t flag = *const_cast<uint8_t *>(baseBuf);
1004
1005 switch (flag) {
1006 case SP_JDWP_NEWFD:
1007 case SP_ARK_NEWFD: {
1008 JdwpNewFileDescriptor(baseBuf, bytesIO);
1009 break;
1010 }
1011 default:
1012 WRITE_LOG(LOG_WARN, "Not support session command");
1013 break;
1014 }
1015 return ret;
1016 }
1017
ReadCtrlFromSession(uv_poll_t * poll,int status,int events)1018 void HdcSessionBase::ReadCtrlFromSession(uv_poll_t *poll, int status, int events)
1019 {
1020 HSession hSession = (HSession)poll->data;
1021 HdcSessionBase *hSessionBase = (HdcSessionBase *)hSession->classInstance;
1022 const int size = Base::GetMaxBufSize();
1023 char *buf = reinterpret_cast<char *>(new uint8_t[size]());
1024 ssize_t nread = Base::ReadFromFd(hSession->ctrlFd[STREAM_MAIN], buf, size);
1025 while (true) {
1026 if (nread < 0) {
1027 constexpr int bufSize = 1024;
1028 char buffer[bufSize] = { 0 };
1029 uv_strerror_r(static_cast<int>(nread), buffer, bufSize);
1030 WRITE_LOG(LOG_DEBUG, "ReadCtrlFromSession failed,%s", buffer);
1031 uv_poll_stop(poll);
1032 break;
1033 }
1034 if (nread == 0) {
1035 WRITE_LOG(LOG_FATAL, "ReadCtrlFromSession read data zero byte");
1036 break;
1037 }
1038 // only one command, no need to split command from stream
1039 // if add more commands, consider the format command
1040 hSessionBase->DispatchSessionThreadCommand(hSession, reinterpret_cast<uint8_t *>(buf), nread);
1041 break;
1042 }
1043 delete[] buf;
1044 }
1045
WorkThreadStartSession(HSession hSession)1046 bool HdcSessionBase::WorkThreadStartSession(HSession hSession)
1047 {
1048 bool regOK = false;
1049 int childRet = 0;
1050 if (hSession->connType == CONN_TCP) {
1051 HdcTCPBase *pTCPBase = (HdcTCPBase *)hSession->classModule;
1052 hSession->hChildWorkTCP.data = hSession;
1053 if (uv_tcp_init(&hSession->childLoop, &hSession->hChildWorkTCP) < 0) {
1054 WRITE_LOG(LOG_DEBUG, "HdcSessionBase SessionCtrl failed 1");
1055 return false;
1056 }
1057 if ((childRet = uv_tcp_open(&hSession->hChildWorkTCP, hSession->fdChildWorkTCP)) < 0) {
1058 constexpr int bufSize = 1024;
1059 char buf[bufSize] = { 0 };
1060 uv_strerror_r(childRet, buf, bufSize);
1061 WRITE_LOG(LOG_DEBUG, "SessionCtrl failed 2,fd:%d,str:%s", hSession->fdChildWorkTCP, buf);
1062 return false;
1063 }
1064 Base::SetTcpOptions((uv_tcp_t *)&hSession->hChildWorkTCP);
1065 uv_read_start((uv_stream_t *)&hSession->hChildWorkTCP, AllocCallback, pTCPBase->ReadStream);
1066 regOK = true;
1067 #ifdef HDC_SUPPORT_UART
1068 } else if (hSession->connType == CONN_SERIAL) { // UART
1069 HdcUARTBase *pUARTBase = (HdcUARTBase *)hSession->classModule;
1070 WRITE_LOG(LOG_DEBUG, "UART ReadyForWorkThread");
1071 regOK = pUARTBase->ReadyForWorkThread(hSession);
1072 #endif
1073 } else { // USB
1074 HdcUSBBase *pUSBBase = (HdcUSBBase *)hSession->classModule;
1075 WRITE_LOG(LOG_DEBUG, "USB ReadyForWorkThread");
1076 regOK = pUSBBase->ReadyForWorkThread(hSession);
1077 }
1078
1079 if (regOK && hSession->serverOrDaemon) {
1080 // session handshake step1
1081 SessionHandShake handshake = {};
1082 handshake.banner = HANDSHAKE_MESSAGE;
1083 handshake.sessionId = hSession->sessionId;
1084 handshake.connectKey = hSession->connectKey;
1085 if (!hSession->isCheck) {
1086 handshake.version = Base::GetVersion() + HDC_MSG_HASH;
1087 WRITE_LOG(LOG_INFO, "set version = %s", handshake.version.c_str());
1088 }
1089 handshake.authType = AUTH_NONE;
1090 string hs = SerialStruct::SerializeToString(handshake);
1091 #ifdef HDC_SUPPORT_UART
1092 WRITE_LOG(LOG_DEBUG, "WorkThreadStartSession session %u auth %u send handshake hs: %s",
1093 hSession->sessionId, handshake.authType, hs.c_str());
1094 #endif
1095 Send(hSession->sessionId, 0, CMD_KERNEL_HANDSHAKE,
1096 reinterpret_cast<uint8_t *>(const_cast<char *>(hs.c_str())), hs.size());
1097 }
1098 return regOK;
1099 }
1100
BuildCtrlString(InnerCtrlCommand command,uint32_t channelId,uint8_t * data,int dataSize)1101 vector<uint8_t> HdcSessionBase::BuildCtrlString(InnerCtrlCommand command, uint32_t channelId, uint8_t *data,
1102 int dataSize)
1103 {
1104 vector<uint8_t> ret;
1105 while (true) {
1106 if (dataSize > BUF_SIZE_MICRO) {
1107 WRITE_LOG(LOG_WARN, "BuildCtrlString dataSize:%d", dataSize);
1108 break;
1109 }
1110 CtrlStruct ctrl = {};
1111 ctrl.command = command;
1112 ctrl.channelId = channelId;
1113 ctrl.dataSize = dataSize;
1114 if (dataSize > 0 && data != nullptr && memcpy_s(ctrl.data, sizeof(ctrl.data), data, dataSize) != EOK) {
1115 break;
1116 }
1117 uint8_t *buf = reinterpret_cast<uint8_t *>(&ctrl);
1118 ret.insert(ret.end(), buf, buf + sizeof(CtrlStruct));
1119 break;
1120 }
1121 return ret;
1122 }
1123
DispatchMainThreadCommand(HSession hSession,const CtrlStruct * ctrl)1124 bool HdcSessionBase::DispatchMainThreadCommand(HSession hSession, const CtrlStruct *ctrl)
1125 {
1126 bool ret = true;
1127 uint32_t channelId = ctrl->channelId; // if send not set, it is zero
1128 switch (ctrl->command) {
1129 case SP_START_SESSION: {
1130 WRITE_LOG(LOG_DEBUG, "Dispatch MainThreadCommand START_SESSION sessionId:%u instance:%s",
1131 hSession->sessionId, hSession->serverOrDaemon ? "server" : "daemon");
1132 ret = WorkThreadStartSession(hSession);
1133 break;
1134 }
1135 case SP_STOP_SESSION: {
1136 WRITE_LOG(LOG_DEBUG, "Dispatch MainThreadCommand STOP_SESSION sessionId:%u", hSession->sessionId);
1137 auto closeSessionChildThreadTCPHandle = [](uv_handle_t *handle) -> void {
1138 HSession hSession = (HSession)handle->data;
1139 Base::TryCloseHandle((uv_handle_t *)handle);
1140 if (handle == (uv_handle_t *)hSession->pollHandle[STREAM_WORK]) {
1141 free(hSession->pollHandle[STREAM_WORK]);
1142 }
1143 if (--hSession->uvChildRef == 0) {
1144 uv_stop(&hSession->childLoop);
1145 };
1146 };
1147 constexpr int uvChildRefOffset = 2;
1148 hSession->uvChildRef += uvChildRefOffset;
1149 if (hSession->connType == CONN_TCP && hSession->hChildWorkTCP.loop) { // maybe not use it
1150 ++hSession->uvChildRef;
1151 Base::TryCloseHandle((uv_handle_t *)&hSession->hChildWorkTCP, true, closeSessionChildThreadTCPHandle);
1152 }
1153 Base::TryCloseHandle((uv_handle_t *)hSession->pollHandle[STREAM_WORK], true,
1154 closeSessionChildThreadTCPHandle);
1155 Base::TryCloseHandle((uv_handle_t *)&hSession->dataPipe[STREAM_WORK], true,
1156 closeSessionChildThreadTCPHandle);
1157 break;
1158 }
1159 case SP_ATTACH_CHANNEL: {
1160 if (!serverOrDaemon) {
1161 break; // Only Server has this feature
1162 }
1163 AttachChannel(hSession, channelId);
1164 break;
1165 }
1166 case SP_DEATCH_CHANNEL: {
1167 if (!serverOrDaemon) {
1168 break; // Only Server has this feature
1169 }
1170 DeatchChannel(hSession, channelId);
1171 break;
1172 }
1173 default:
1174 WRITE_LOG(LOG_WARN, "Not support main command");
1175 ret = false;
1176 break;
1177 }
1178 return ret;
1179 }
1180
1181 // Several bytes of control instructions, generally do not stick
ReadCtrlFromMain(uv_poll_t * poll,int status,int events)1182 void HdcSessionBase::ReadCtrlFromMain(uv_poll_t *poll, int status, int events)
1183 {
1184 HSession hSession = (HSession)poll->data;
1185 HdcSessionBase *hSessionBase = (HdcSessionBase *)hSession->classInstance;
1186 int formatCommandSize = sizeof(CtrlStruct);
1187 int index = 0;
1188 const int size = Base::GetMaxBufSize();
1189 char *buf = reinterpret_cast<char *>(new uint8_t[size]());
1190 ssize_t nread = Base::ReadFromFd(hSession->ctrlFd[STREAM_WORK], buf, size);
1191 while (true) {
1192 if (nread < 0) {
1193 constexpr int bufSize = 1024;
1194 char buffer[bufSize] = { 0 };
1195 uv_strerror_r(static_cast<int>(nread), buffer, bufSize);
1196 WRITE_LOG(LOG_DEBUG, "SessionCtrl failed,%s", buffer);
1197 break;
1198 }
1199 if (nread % formatCommandSize != 0) {
1200 WRITE_LOG(LOG_FATAL, "ReadCtrlFromMain size failed, nread == %d", nread);
1201 break;
1202 }
1203 CtrlStruct *ctrl = reinterpret_cast<CtrlStruct *>(buf + index);
1204 if (!hSessionBase->DispatchMainThreadCommand(hSession, ctrl)) {
1205 WRITE_LOG(LOG_FATAL, "ReadCtrlFromMain failed sessionId:%u channelId:%u command:%u",
1206 hSession->sessionId, ctrl->channelId, ctrl->command);
1207 break;
1208 }
1209 index += sizeof(CtrlStruct);
1210 if (index >= nread) {
1211 break;
1212 }
1213 }
1214 delete[] buf;
1215 }
1216
ReChildLoopForSessionClear(HSession hSession)1217 void HdcSessionBase::ReChildLoopForSessionClear(HSession hSession)
1218 {
1219 // Restart loop close task
1220 ClearOwnTasks(hSession, 0);
1221 WRITE_LOG(LOG_INFO, "ReChildLoopForSessionClear sessionId:%u", hSession->sessionId);
1222 auto clearTaskForSessionFinish = [](uv_timer_t *handle) -> void {
1223 HSession hSession = (HSession)handle->data;
1224 for (auto v : *hSession->mapTask) {
1225 HTaskInfo hTask = (HTaskInfo)v.second;
1226 uint8_t level;
1227 if (hTask->closeRetryCount < GLOBAL_TIMEOUT / 2) {
1228 level = LOG_DEBUG;
1229 } else {
1230 level = LOG_WARN;
1231 }
1232 WRITE_LOG(level, "wait task free retry %d/%d, channelId:%u, sessionId:%u",
1233 hTask->closeRetryCount, GLOBAL_TIMEOUT, hTask->channelId, hTask->sessionId);
1234 if (hTask->closeRetryCount++ >= GLOBAL_TIMEOUT) {
1235 HdcSessionBase *thisClass = (HdcSessionBase *)hTask->ownerSessionClass;
1236 hSession = thisClass->AdminSession(OP_QUERY, hTask->sessionId, nullptr);
1237 thisClass->AdminTask(OP_VOTE_RESET, hSession, hTask->channelId, nullptr);
1238 }
1239 if (!hTask->taskFree)
1240 return;
1241 }
1242 // all task has been free
1243 uv_close((uv_handle_t *)handle, Base::CloseTimerCallback);
1244 uv_stop(&hSession->childLoop); // stop ReChildLoopForSessionClear pendding
1245 };
1246 Base::TimerUvTask(
1247 &hSession->childLoop, hSession, clearTaskForSessionFinish, (GLOBAL_TIMEOUT * TIME_BASE) / UV_DEFAULT_INTERVAL);
1248 uv_run(&hSession->childLoop, UV_RUN_DEFAULT);
1249 // clear
1250 Base::TryCloseChildLoop(&hSession->childLoop, "Session childUV");
1251 }
1252
SessionWorkThread(uv_work_t * arg)1253 void HdcSessionBase::SessionWorkThread(uv_work_t *arg)
1254 {
1255 HSession hSession = (HSession)arg->data;
1256 HdcSessionBase *thisClass = (HdcSessionBase *)hSession->classInstance;
1257 hSession->hWorkChildThread = uv_thread_self();
1258
1259 uv_poll_t *pollHandle = hSession->pollHandle[STREAM_WORK];
1260 pollHandle->data = hSession;
1261 uv_poll_init_socket(&hSession->childLoop, pollHandle, hSession->ctrlFd[STREAM_WORK]);
1262 uv_poll_start(pollHandle, UV_READABLE, ReadCtrlFromMain);
1263 WRITE_LOG(LOG_DEBUG, "!!!Workthread run begin, sessionId:%u instance:%s", hSession->sessionId,
1264 thisClass->serverOrDaemon ? "server" : "daemon");
1265 uv_run(&hSession->childLoop, UV_RUN_DEFAULT); // work pendding
1266 WRITE_LOG(LOG_DEBUG, "!!!Workthread run again, sessionId:%u", hSession->sessionId);
1267 // main loop has exit
1268 thisClass->ReChildLoopForSessionClear(hSession); // work pending again
1269 hSession->childCleared = true;
1270 WRITE_LOG(LOG_DEBUG, "!!!Workthread run finish, sessionId:%u", hSession->sessionId);
1271 }
1272
1273 // clang-format off
LogMsg(const uint32_t sessionId,const uint32_t channelId,MessageLevel level,const char * msg,...)1274 void HdcSessionBase::LogMsg(const uint32_t sessionId, const uint32_t channelId,
1275 MessageLevel level, const char *msg, ...)
1276 // clang-format on
1277 {
1278 va_list vaArgs;
1279 va_start(vaArgs, msg);
1280 string log = Base::StringFormat(msg, vaArgs);
1281 va_end(vaArgs);
1282 vector<uint8_t> buf;
1283 buf.push_back(level);
1284 buf.insert(buf.end(), log.c_str(), log.c_str() + log.size());
1285 ServerCommand(sessionId, channelId, CMD_KERNEL_ECHO, buf.data(), buf.size());
1286 }
1287
NeedNewTaskInfo(const uint16_t command,bool & masterTask)1288 bool HdcSessionBase::NeedNewTaskInfo(const uint16_t command, bool &masterTask)
1289 {
1290 // referer from HdcServerForClient::DoCommandRemote
1291 bool ret = false;
1292 bool taskMasterInit = false;
1293 masterTask = false;
1294 switch (command) {
1295 case CMD_FILE_INIT:
1296 case CMD_FLASHD_FLASH_INIT:
1297 case CMD_FLASHD_UPDATE_INIT:
1298 case CMD_FLASHD_ERASE:
1299 case CMD_FLASHD_FORMAT:
1300 case CMD_FORWARD_INIT:
1301 case CMD_APP_INIT:
1302 case CMD_APP_UNINSTALL:
1303 case CMD_APP_SIDELOAD:
1304 taskMasterInit = true;
1305 break;
1306 default:
1307 break;
1308 }
1309 if (!serverOrDaemon
1310 && (command == CMD_SHELL_INIT || (command > CMD_UNITY_COMMAND_HEAD && command < CMD_UNITY_COMMAND_TAIL))) {
1311 // daemon's single side command
1312 ret = true;
1313 } else if (command == CMD_KERNEL_WAKEUP_SLAVETASK) {
1314 // slave tasks
1315 ret = true;
1316 } else if (command == CMD_UNITY_BUGREPORT_INIT) {
1317 ret = true;
1318 } else if (taskMasterInit) {
1319 // task init command
1320 masterTask = true;
1321 ret = true;
1322 }
1323 return ret;
1324 }
1325 // Heavy and time-consuming work was putted in the new thread to do, and does
1326 // not occupy the main thread
DispatchTaskData(HSession hSession,const uint32_t channelId,const uint16_t command,uint8_t * payload,int payloadSize)1327 bool HdcSessionBase::DispatchTaskData(HSession hSession, const uint32_t channelId, const uint16_t command,
1328 uint8_t *payload, int payloadSize)
1329 {
1330 StartTraceScope("HdcSessionBase::DispatchTaskData");
1331 bool ret = true;
1332 HTaskInfo hTaskInfo = nullptr;
1333 bool masterTask = false;
1334 while (true) {
1335 // Some basic commands do not have a local task constructor. example: Interactive shell, some uinty commands
1336 if (NeedNewTaskInfo(command, masterTask)) {
1337 WRITE_LOG(LOG_DEBUG, "New HTaskInfo channelId:%u command:%u", channelId, command);
1338 hTaskInfo = new(std::nothrow) TaskInformation();
1339 if (hTaskInfo == nullptr) {
1340 WRITE_LOG(LOG_FATAL, "DispatchTaskData new hTaskInfo failed");
1341 break;
1342 }
1343 hTaskInfo->channelId = channelId;
1344 hTaskInfo->sessionId = hSession->sessionId;
1345 hTaskInfo->runLoop = &hSession->childLoop;
1346 hTaskInfo->serverOrDaemon = serverOrDaemon;
1347 hTaskInfo->masterSlave = masterTask;
1348 hTaskInfo->closeRetryCount = 0;
1349 hTaskInfo->channelTask = false;
1350
1351 int addTaskRetry = 3; // try 3 time
1352 while (addTaskRetry > 0) {
1353 if (AdminTask(OP_ADD, hSession, channelId, hTaskInfo)) {
1354 break;
1355 }
1356 sleep(1);
1357 --addTaskRetry;
1358 }
1359
1360 if (addTaskRetry == 0) {
1361 #ifndef HDC_HOST
1362 LogMsg(hTaskInfo->sessionId, hTaskInfo->channelId,
1363 MSG_FAIL, "hdc thread pool busy, may cause reset later");
1364 #endif
1365 delete hTaskInfo;
1366 hTaskInfo = nullptr;
1367 ret = false;
1368 break;
1369 }
1370 } else {
1371 hTaskInfo = AdminTask(OP_QUERY, hSession, channelId, nullptr);
1372 }
1373 if (!hTaskInfo || hTaskInfo->taskStop || hTaskInfo->taskFree) {
1374 WRITE_LOG(LOG_ALL, "Dead HTaskInfo, ignore, channelId:%u command:%u", channelId, command);
1375 break;
1376 }
1377 ret = RedirectToTask(hTaskInfo, hSession, channelId, command, payload, payloadSize);
1378 break;
1379 }
1380 return ret;
1381 }
1382
PostStopInstanceMessage(bool restart)1383 void HdcSessionBase::PostStopInstanceMessage(bool restart)
1384 {
1385 PushAsyncMessage(0, ASYNC_STOP_MAINLOOP, nullptr, 0);
1386 WRITE_LOG(LOG_DEBUG, "StopDaemon has sended restart %d", restart);
1387 wantRestart = restart;
1388 }
1389 } // namespace Hdc
1390