1/*
2 Copyright (C) 2012 Samsung Electronics
3
4 This library is free software; you can redistribute it and/or
5 modify it under the terms of the GNU Library General Public
6 License as published by the Free Software Foundation; either
7 version 2 of the License, or (at your option) any later version.
8
9 This library 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 GNU
12 Library General Public License for more details.
13
14 You should have received a copy of the GNU Library General Public License
15 along with this library; see the file COPYING.LIB. If not, write to
16 the Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor,
17 Boston, MA 02110-1301, USA.
18 */
19
20#include "config.h"
21#include "WorkQueue.h"
22
23#include <wtf/Assertions.h>
24
25class WorkQueue::TimerWorkItem {
26public:
27 TimerWorkItem(int timerID, const Function<void()>& function, WorkQueue* queue)
28 : m_function(function)
29 , m_queue(queue)
30 , m_timerID(timerID)
31 {
32 }
33 ~TimerWorkItem() { }
34
35 Function<void()> function() const { return m_function; }
36 WorkQueue* queue() const { return m_queue; }
37
38 int timerID() const { return m_timerID; }
39
40private:
41 Function<void()> m_function;
42 WorkQueue* m_queue;
43 int m_timerID;
44};
45
46static const int invalidSocketDescriptor = -1;
47static const int threadMessageSize = 1;
48static const char* const finishThreadMessage = "F";
49static const char* const wakupThreadMessage = "W";
50
51// WorkQueue
52void WorkQueue::platformInitialize(const char* name)
53{
54 int fds[2];
55 if (pipe(fds))
56 ASSERT_NOT_REACHED();
57 m_readFromPipeFd = fds[0];
58 m_writeToPipeFd = fds[1];
59 FD_ZERO(&m_fdSet);
60 FD_SET(m_readFromPipeFd, &m_fdSet);
61 m_maxFd = m_readFromPipeFd;
62
63 m_socketDescriptor = invalidSocketDescriptor;
64
65 m_processExitHandler = ecore_event_handler_add(ECORE_EXE_EVENT_DEL, reinterpret_cast<Ecore_Event_Handler_Cb>(processExitCallBack), this);
66
67 m_threadLoop = true;
68 createThread(reinterpret_cast<WTF::ThreadFunction>(&WorkQueue::workQueueThread), this, name);
69}
70
71void WorkQueue::platformInvalidate()
72{
73 if (m_processExitHandler)
74 ecore_event_handler_del(m_processExitHandler);
75 m_processExitHandler = 0;
76
77 sendMessageToThread(finishThreadMessage);
78}
79
80void WorkQueue::performWork()
81{
82 m_workItemQueueLock.lock();
83
84 while (!m_workItemQueue.isEmpty()) {
85 Vector<Function<void()> > workItemQueue;
86 m_workItemQueue.swap(workItemQueue);
87
88 m_workItemQueueLock.unlock();
89 for (size_t i = 0; i < workItemQueue.size(); ++i)
90 workItemQueue[i]();
91 m_workItemQueueLock.lock();
92 }
93 m_workItemQueueLock.unlock();
94}
95
96void WorkQueue::performFdWork()
97{
98 fd_set readFdSet = m_fdSet;
99
100 if (select(m_maxFd + 1, &readFdSet, 0, 0, 0) >= 0) {
101 if (FD_ISSET(m_readFromPipeFd, &readFdSet)) {
102 char readBuf[threadMessageSize];
103 if (read(m_readFromPipeFd, readBuf, threadMessageSize) == -1)
104 LOG_ERROR("Failed to read from WorkQueueThread pipe");
105 if (!strncmp(readBuf, finishThreadMessage, threadMessageSize))
106 m_threadLoop = false;
107 }
108
109 if (m_socketDescriptor != invalidSocketDescriptor && FD_ISSET(m_socketDescriptor, &readFdSet))
110 m_socketEventHandler();
111 }
112}
113
114void WorkQueue::sendMessageToThread(const char* message)
115{
116 if (write(m_writeToPipeFd, message, threadMessageSize) == -1)
117 LOG_ERROR("Failed to wake up WorkQueue Thread");
118}
119
120void* WorkQueue::workQueueThread(WorkQueue* workQueue)
121{
122 while (workQueue->m_threadLoop) {
123 workQueue->performWork();
124 workQueue->performFdWork();
125 }
126
127 close(workQueue->m_readFromPipeFd);
128 close(workQueue->m_writeToPipeFd);
129
130 return 0;
131}
132
133void WorkQueue::registerSocketEventHandler(int fd, const Function<void()>& function)
134{
135 if (m_socketDescriptor != invalidSocketDescriptor)
136 LOG_ERROR("%d is already registerd.", fd);
137
138 m_socketDescriptor = fd;
139 m_socketEventHandler = function;
140
141 if (fd > m_maxFd)
142 m_maxFd = fd;
143 FD_SET(fd, &m_fdSet);
144}
145
146void WorkQueue::unregisterSocketEventHandler(int fd)
147{
148 m_socketDescriptor = invalidSocketDescriptor;
149
150 if (fd == m_maxFd)
151 m_maxFd = m_readFromPipeFd;
152 FD_CLR(fd, &m_fdSet);
153}
154
155void WorkQueue::scheduleWorkAndWakeUp(const Function<void()>& function)
156{
157 if (function.isNull())
158 return;
159
160 MutexLocker locker(m_workItemQueueLock);
161 m_workItemQueue.append(function);
162 sendMessageToThread(wakupThreadMessage);
163}
164
165void WorkQueue::dispatch(const Function<void()>& function)
166{
167 scheduleWorkAndWakeUp(function);
168}
169
170bool WorkQueue::timerFired(void* data)
171{
172 TimerWorkItem* item = static_cast<TimerWorkItem*>(data);
173 if (item && item->queue()->m_isValid) {
174 item->queue()->scheduleWorkAndWakeUp(item->function());
175 item->queue()->m_timers.take(item->timerID());
176 delete item;
177 }
178
179 return ECORE_CALLBACK_CANCEL;
180}
181
182void WorkQueue::dispatchAfterDelay(const Function<void()>& function, double delay)
183{
184 static int timerId = 0;
185 m_timers.set(timerId, adoptPtr(ecore_timer_add(delay, reinterpret_cast<Ecore_Task_Cb>(timerFired), new TimerWorkItem(timerId, function, this))));
186 timerId++;
187}
188
189bool WorkQueue::processExitCallBack(void* data, int type, void* event)
190{
191 WorkQueue* queue = reinterpret_cast<WorkQueue*>(data);
192 if (queue->m_isValid) {
193 Ecore_Exe_Event_Del* ev = static_cast<Ecore_Exe_Event_Del*>(event);
194 WebKit::PlatformProcessIdentifier deadChildPid = static_cast<WebKit::PlatformProcessIdentifier>(ev->pid);
195
196 if (queue->m_termWorkItems.contains(deadChildPid)) {
197 Function<void()> function = queue->m_termWorkItems.take(deadChildPid);
198 queue->scheduleWorkAndWakeUp(function);
199 }
200 }
201 return ECORE_CALLBACK_RENEW;
202}
203
204void WorkQueue::dispatchOnTermination(WebKit::PlatformProcessIdentifier childPid, const Function<void()>& function)
205{
206 ASSERT_ARG(childPid, !m_termWorkItems.contains(childPid));
207 m_termWorkItems.set(childPid, function);
208}