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 
16 #include "nstackx_dfile_mp.h"
17 #include "nstackx_dfile_log.h"
18 #include "nstackx_dfile_send.h"
19 #include "nstackx_dfile_transfer.h"
20 #include "nstackx_file_manager.h"
21 #include "securec.h"
22 
23 #define TAG "nStackXDfileMp"
24 
DFileSocketRecvSP(DFileSession * session)25 int32_t DFileSocketRecvSP(DFileSession *session)
26 {
27     struct sockaddr_in peerAddr;
28     socklen_t addrLen = sizeof(struct sockaddr_in);
29     uint8_t frame[NSTACKX_MAX_FRAME_SIZE] = {0};
30     int32_t ret;
31 
32     (void)memset_s(&peerAddr, addrLen, 0, addrLen);
33     if (CapsTcp(session)) {
34         if (session->sessionType == DFILE_SESSION_TYPE_SERVER) {
35             if (session->acceptFlag == 0) {
36                 ret = DFileAcceptSocket(session);
37                 return ret;
38             } else {
39                 ret = SocketRecvForTcp(session, session->recvBuffer, &peerAddr, &addrLen);
40             }
41         } else {
42             ret = SocketRecvForTcp(session, session->recvBuffer, &peerAddr, &addrLen);
43         }
44         if (ret == NSTACKX_PEER_CLOSE) {
45             return ret;
46         }
47     } else {
48         ret = SocketRecv(session->socket[0], frame, sizeof(frame), &peerAddr, &addrLen);
49     }
50     if (ret <= 0) {
51         if (ret != NSTACKX_EAGAIN) {
52             DFILE_LOGE(TAG, "socket recv failed");
53             return NSTACKX_EFAILED;
54         }
55         return NSTACKX_EAGAIN;
56     }
57     NSTACKX_ATOM_FETCH_INC(&session->totalRecvBlocks);
58     if (CapsTcp(session)) {
59         ret = DFileSessionHandleReadBuffer(session, session->recvBuffer, (size_t)session->recvLen, &peerAddr, 0);
60         session->recvLen = 0;
61     } else {
62         ret = DFileSessionHandleReadBuffer(session, frame, (size_t)ret, &peerAddr, 0);
63     }
64     if (ret != NSTACKX_EOK) {
65         DFILE_LOGE(TAG, "handle read buffer failed");
66     }
67     return ret;
68 }
69 
TransSelectPeerInfo(DFileSession * session)70 PeerInfo *TransSelectPeerInfo(DFileSession *session)
71 {
72     PeerInfo *peerInfo = (PeerInfo *)ListGetFront(&session->peerInfoChain);
73     return peerInfo;
74 }
75 
76 /* only for client */
ClientGetPeerInfoByTransId(DFileSession * session)77 PeerInfo *ClientGetPeerInfoByTransId(DFileSession *session)
78 {
79     PeerInfo *peerInfo = (PeerInfo *)ListGetFront(&session->peerInfoChain);
80     return peerInfo;
81 }
82 
83 /* only for client */
ClientGetPeerInfoBySocketIndex(uint8_t socketIndex,const DFileSession * session)84 PeerInfo *ClientGetPeerInfoBySocketIndex(uint8_t socketIndex, const DFileSession *session)
85 {
86     PeerInfo *peerInfo = NULL;
87     List     *pos = NULL;
88 
89     LIST_FOR_EACH(pos, &session->peerInfoChain) {
90         peerInfo = (PeerInfo *)pos;
91         if (peerInfo->socketIndex == socketIndex) {
92             return peerInfo;
93         }
94     }
95     return NULL;
96 }
97 
CreateSenderThread(DFileSession * session)98 int32_t CreateSenderThread(DFileSession *session)
99 {
100     SenderThreadPara *para = NULL;
101 
102     para = malloc(sizeof(SenderThreadPara));
103     if (para == NULL) {
104         DFILE_LOGE(TAG, "Failed to allocate memory for SenderThreadPara");
105         return NSTACKX_ENOMEM;
106     }
107     para->session = session;
108     para->socketIndex = 0;
109     if (PthreadCreate(&(session->senderTid[0]), NULL, DFileSenderHandle, para)) {
110         DFILE_LOGE(TAG, "Create sender thread 0 failed");
111         free(para);
112         return NSTACKX_EFAILED;
113     }
114 
115     return NSTACKX_EOK;
116 }
117 
RebuildFilelist(const char * files[],const char * remotePath[],uint32_t fileNum,DFileSession * session,DFileRebuildFileList * rebuildList)118 int32_t RebuildFilelist(const char *files[], const char *remotePath[], uint32_t fileNum,
119     DFileSession *session, DFileRebuildFileList *rebuildList)
120 {
121     if (session->allTaskCount >= NSTACKX_MAX_FILE_LIST_NUM) {
122         DFILE_LOGI(TAG, "more than %d send task", NSTACKX_MAX_FILE_LIST_NUM);
123         return NSTACKX_EFAILED;
124     }
125 
126     (void)memset_s(rebuildList, sizeof(DFileRebuildFileList), 0, sizeof(DFileRebuildFileList));
127 
128     rebuildList->transNum = 1;
129     for (uint32_t i = 0; i < fileNum; i++) {
130         rebuildList->files[i] = files[i];
131         if (remotePath) {
132             rebuildList->remotePath[i] = remotePath[i];
133         }
134     }
135     return NSTACKX_EOK;
136 }
137 
InitOutboundQueueWait(DFileSession * session)138 int32_t InitOutboundQueueWait(DFileSession *session)
139 {
140     if (SemInit(&session->outboundQueueWait[0], 0, 0) != 0) {
141         return NSTACKX_EFAILED;
142     }
143 
144     return NSTACKX_EOK;
145 }
146 
DestroyOutboundQueueWait(DFileSession * session)147 void DestroyOutboundQueueWait(DFileSession *session)
148 {
149     SemDestroy(&session->outboundQueueWait[0]);
150 }
151 
PostOutboundQueueWait(DFileSession * session)152 void PostOutboundQueueWait(DFileSession *session)
153 {
154     SemPost(&session->outboundQueueWait[0]);
155 }
156