rsThreadIO.cpp revision 6b0c00414a532e2ea22f66cf89578cc8c8a45293 
1/*
2 * Copyright (C) 2009 The Android Open Source Project
3 *
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
7 *
8 *      http://www.apache.org/licenses/LICENSE-2.0
9 *
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
15 */
16
17#include "rsContext.h"
18
19#include "rsThreadIO.h"
20
21using namespace android;
22using namespace android::renderscript;
23
24ThreadIO::ThreadIO() : mUsingSocket(false) {
25}
26
27ThreadIO::~ThreadIO() {
28}
29
30void ThreadIO::init(bool useSocket) {
31    mUsingSocket = useSocket;
32    mToCore.init(16 * 1024);
33
34    if (mUsingSocket) {
35        mToClientSocket.init();
36        mToCoreSocket.init();
37    } else {
38        mToClient.init(1024);
39    }
40}
41
42void ThreadIO::shutdown() {
43    //LOGE("shutdown 1");
44    mToCore.shutdown();
45    //LOGE("shutdown 2");
46}
47
48void ThreadIO::coreFlush() {
49    //LOGE("coreFlush 1");
50    if (mUsingSocket) {
51    } else {
52        mToCore.flush();
53    }
54    //LOGE("coreFlush 2");
55}
56
57void * ThreadIO::coreHeader(uint32_t cmdID, size_t dataLen) {
58    //LOGE("coreHeader %i %i", cmdID, dataLen);
59    if (mUsingSocket) {
60        CoreCmdHeader hdr;
61        hdr.bytes = dataLen;
62        hdr.cmdID = cmdID;
63        mToCoreSocket.writeAsync(&hdr, sizeof(hdr));
64    } else {
65        mCoreCommandSize = dataLen;
66        mCoreCommandID = cmdID;
67        mCoreDataPtr = (uint8_t *)mToCore.reserve(dataLen);
68        mCoreDataBasePtr = mCoreDataPtr;
69    }
70    //LOGE("coreHeader ret %p", mCoreDataPtr);
71    return mCoreDataPtr;
72}
73
74void ThreadIO::coreData(const void *data, size_t dataLen) {
75    //LOGE("coreData %p %i", data, dataLen);
76    mToCoreSocket.writeAsync(data, dataLen);
77    //LOGE("coreData ret %p", mCoreDataPtr);
78}
79
80void ThreadIO::coreCommit() {
81    //LOGE("coreCommit %p %p %i", mCoreDataPtr, mCoreDataBasePtr, mCoreCommandSize);
82    if (mUsingSocket) {
83    } else {
84        rsAssert((size_t)(mCoreDataPtr - mCoreDataBasePtr) <= mCoreCommandSize);
85        mToCore.commit(mCoreCommandID, mCoreCommandSize);
86    }
87    //LOGE("coreCommit ret");
88}
89
90void ThreadIO::coreCommitSync() {
91    //LOGE("coreCommitSync %p %p %i", mCoreDataPtr, mCoreDataBasePtr, mCoreCommandSize);
92    if (mUsingSocket) {
93    } else {
94        rsAssert((size_t)(mCoreDataPtr - mCoreDataBasePtr) <= mCoreCommandSize);
95        mToCore.commitSync(mCoreCommandID, mCoreCommandSize);
96    }
97    //LOGE("coreCommitSync ret");
98}
99
100void ThreadIO::clientShutdown() {
101    //LOGE("coreShutdown 1");
102    mToClient.shutdown();
103    //LOGE("coreShutdown 2");
104}
105
106void ThreadIO::coreSetReturn(const void *data, size_t dataLen) {
107    rsAssert(dataLen <= sizeof(mToCoreRet));
108    memcpy(&mToCoreRet, data, dataLen);
109}
110
111void ThreadIO::coreGetReturn(void *data, size_t dataLen) {
112    memcpy(data, &mToCoreRet, dataLen);
113}
114
115
116bool ThreadIO::playCoreCommands(Context *con, bool waitForCommand) {
117    bool ret = false;
118    while (!mToCore.isEmpty() || waitForCommand) {
119        uint32_t cmdID = 0;
120        uint32_t cmdSize = 0;
121        ret = true;
122        if (con->props.mLogTimes) {
123            con->timerSet(Context::RS_TIMER_IDLE);
124        }
125        const void * data = mToCore.get(&cmdID, &cmdSize);
126        if (!cmdSize) {
127            // exception occured, probably shutdown.
128            return false;
129        }
130        if (con->props.mLogTimes) {
131            con->timerSet(Context::RS_TIMER_INTERNAL);
132        }
133        waitForCommand = false;
134        //LOGV("playCoreCommands 3 %i %i", cmdID, cmdSize);
135
136        if (cmdID >= (sizeof(gPlaybackFuncs) / sizeof(void *))) {
137            rsAssert(cmdID < (sizeof(gPlaybackFuncs) / sizeof(void *)));
138            LOGE("playCoreCommands error con %p, cmd %i", con, cmdID);
139            mToCore.printDebugData();
140        }
141        gPlaybackFuncs[cmdID](con, data, cmdSize << 2);
142        mToCore.next();
143    }
144    return ret;
145}
146
147RsMessageToClientType ThreadIO::getClientHeader(size_t *receiveLen, uint32_t *usrID) {
148    if (mUsingSocket) {
149        mToClientSocket.read(&mLastClientHeader, sizeof(mLastClientHeader));
150    } else {
151        size_t bytesData = 0;
152        const uint32_t *d = (const uint32_t *)mToClient.get(&mLastClientHeader.cmdID, (uint32_t*)&bytesData);
153        if (bytesData >= sizeof(uint32_t)) {
154            mLastClientHeader.userID = d[0];
155            mLastClientHeader.bytes = bytesData - sizeof(uint32_t);
156        } else {
157            mLastClientHeader.userID = 0;
158            mLastClientHeader.bytes = 0;
159        }
160    }
161    receiveLen[0] = mLastClientHeader.bytes;
162    usrID[0] = mLastClientHeader.userID;
163    return (RsMessageToClientType)mLastClientHeader.cmdID;
164}
165
166RsMessageToClientType ThreadIO::getClientPayload(void *data, size_t *receiveLen,
167                                uint32_t *usrID, size_t bufferLen) {
168    receiveLen[0] = mLastClientHeader.bytes;
169    usrID[0] = mLastClientHeader.userID;
170    if (bufferLen < mLastClientHeader.bytes) {
171        return RS_MESSAGE_TO_CLIENT_RESIZE;
172    }
173    if (mUsingSocket) {
174        if (receiveLen[0]) {
175            mToClientSocket.read(data, receiveLen[0]);
176        }
177        return (RsMessageToClientType)mLastClientHeader.cmdID;
178    } else {
179        uint32_t bytesData = 0;
180        uint32_t commandID = 0;
181        const uint32_t *d = (const uint32_t *)mToClient.get(&commandID, &bytesData);
182        //LOGE("getMessageToClient 3    %i  %i", commandID, bytesData);
183        //LOGE("getMessageToClient  %i %i", commandID, *subID);
184        if (bufferLen >= receiveLen[0]) {
185            memcpy(data, d+1, receiveLen[0]);
186            mToClient.next();
187            return (RsMessageToClientType)commandID;
188        }
189    }
190    return RS_MESSAGE_TO_CLIENT_RESIZE;
191}
192
193bool ThreadIO::sendToClient(RsMessageToClientType cmdID, uint32_t usrID, const void *data,
194                            size_t dataLen, bool waitForSpace) {
195    ClientCmdHeader hdr;
196    hdr.bytes = dataLen;
197    hdr.cmdID = cmdID;
198    hdr.userID = usrID;
199    if (mUsingSocket) {
200        mToClientSocket.writeAsync(&hdr, sizeof(hdr));
201        if (dataLen) {
202            mToClientSocket.writeAsync(data, dataLen);
203        }
204        return true;
205    } else {
206        if (!waitForSpace) {
207            if (!mToClient.makeSpaceNonBlocking(dataLen + sizeof(hdr))) {
208                // Not enough room, and not waiting.
209                return false;
210            }
211        }
212
213        //LOGE("sendMessageToClient 2");
214        uint32_t *p = (uint32_t *)mToClient.reserve(dataLen + sizeof(usrID));
215        p[0] = usrID;
216        if (dataLen > 0) {
217            memcpy(p+1, data, dataLen);
218        }
219        mToClient.commit(cmdID, dataLen + sizeof(usrID));
220        //LOGE("sendMessageToClient 3");
221        return true;
222    }
223    return false;
224}
225
226