1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
|
/*******************************************************************************
* Copyright (c) 2012 Ericsson
* Copyright (c) 2010, 2011 École Polytechnique de Montréal
* Copyright (c) 2010, 2011 Alexandre Montplaisir <alexandre.montplaisir@gmail.com>
*
* All rights reserved. This program and the accompanying materials are
* made available under the terms of the Eclipse Public License v1.0 which
* accompanies this distribution, and is available at
* http://www.eclipse.org/legal/epl-v10.html
*
*******************************************************************************/
package org.eclipse.linuxtools.internal.tmf.core.statesystem.backends.historytree;
import java.io.File;
import java.io.IOException;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import org.eclipse.linuxtools.tmf.core.exceptions.TimeRangeException;
import org.eclipse.linuxtools.tmf.core.statevalue.ITmfStateValue;
import org.eclipse.linuxtools.tmf.core.statevalue.TmfStateValue;
import org.eclipse.linuxtools.tmf.core.timestamp.TmfTimestamp;
/**
* Variant of the HistoryTreeBackend which runs all the interval-insertion logic
* in a separate thread.
*
* @author alexmont
*
*/
public final class ThreadedHistoryTreeBackend extends HistoryTreeBackend
implements Runnable {
/*
* From superclass:
*
* protected final StateHistoryTree sht;
*/
private BlockingQueue<HTInterval> intervalQueue;
private final Thread shtThread;
/**
* New state history constructor
*
* Note that it usually doesn't make sense to use a Threaded HT if you're
* opening an existing state-file, but you know what you're doing...
*
* @param newStateFile
* The name of the history file that will be created. Should end
* in ".ht"
* @param blockSize
* The size of the blocks in the file
* @param maxChildren
* The maximum number of children allowed for each core node
* @param startTime
* The earliest timestamp stored in the history
* @param providerVersion
* Version of of the state provider. We will only try to reopen
* existing files if this version matches the one in the
* framework.
* @param queueSize
* The size of the interval insertion queue. 2000 - 10000 usually
* works well
* @throws IOException
* If there was a problem opening the history file for writing
*/
public ThreadedHistoryTreeBackend(File newStateFile, int blockSize,
int maxChildren, long startTime, int providerVersion, int queueSize)
throws IOException {
super(newStateFile, blockSize, maxChildren, providerVersion, startTime);
intervalQueue = new ArrayBlockingQueue<HTInterval>(queueSize);
shtThread = new Thread(this, "History Tree Thread"); //$NON-NLS-1$
shtThread.start();
}
/**
* New State History constructor. This version provides default values for
* blockSize and maxChildren.
*
* @param newStateFile
* The name of the history file that will be created. Should end
* in ".ht"
* @param startTime
* The earliest timestamp stored in the history
* @param providerVersion
* Version of of the state provider. We will only try to reopen
* existing files if this version matches the one in the
* framework.
* @param queueSize
* The size of the interval insertion queue. 2000 - 10000 usually
* works well
* @throws IOException
* If there was a problem opening the history file for writing
*/
public ThreadedHistoryTreeBackend(File newStateFile, long startTime,
int providerVersion, int queueSize) throws IOException {
super(newStateFile, providerVersion, startTime);
intervalQueue = new ArrayBlockingQueue<HTInterval>(queueSize);
shtThread = new Thread(this, "History Tree Thread"); //$NON-NLS-1$
shtThread.start();
}
/*
* The Threaded version does not specify an "existing file" constructor,
* since the history is already built (and we only use the other thread
* during building). Just use a plain HistoryTreeProvider in this case.
*
* TODO but what about streaming??
*/
@Override
public void insertPastState(long stateStartTime, long stateEndTime,
int quark, ITmfStateValue value) throws TimeRangeException {
/*
* Here, instead of directly inserting the elements in the History Tree
* underneath, we'll put them in the Queue. They will then be taken and
* processed by the other thread executing the run() method.
*/
HTInterval interval = new HTInterval(stateStartTime, stateEndTime,
quark, (TmfStateValue) value);
try {
intervalQueue.put(interval);
} catch (InterruptedException e) {
/* We should not get interrupted here */
System.out.println("State system got interrupted!"); //$NON-NLS-1$
e.printStackTrace();
}
}
@Override
public void finishedBuilding(long endTime) {
/*
* We need to commit everything in the History Tree and stop the
* standalone thread before returning to the StateHistorySystem. (SHS
* will then write the Attribute Tree to the file, that must not happen
* at the same time we are writing the last nodes!)
*/
stopRunningThread(endTime);
isFinishedBuilding = true;
return;
}
@Override
public void dispose() {
if (!isFinishedBuilding) {
stopRunningThread(TmfTimestamp.PROJECT_IS_CANNED.getValue());
}
/*
* isFinishedBuilding remains false, so the superclass will ask the
* back-end to delete the file.
*/
super.dispose();
}
private void stopRunningThread(long endTime) {
if (!shtThread.isAlive()) {
return;
}
/*
* Send a "poison pill" in the queue, then wait for the HT to finish
* its closeTree()
*/
try {
HTInterval pill = new HTInterval(-1, endTime, -1, TmfStateValue.nullValue());
intervalQueue.put(pill);
shtThread.join();
} catch (TimeRangeException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
@Override
public void run() {
if (intervalQueue == null) {
System.err.println("Cannot start the storage backend without its interval queue."); //$NON-NLS-1$
return;
}
HTInterval currentInterval;
try {
currentInterval = intervalQueue.take();
while (currentInterval.getStartTime() != -1) {
/* Send the interval to the History Tree */
sht.insertInterval(currentInterval);
currentInterval = intervalQueue.take();
}
assert (currentInterval.getAttribute() == -1);
/*
* We've been told we're done, let's write down everything and quit.
* The end time of this "signal interval" is actually correct.
*/
sht.closeTree(currentInterval.getEndTime());
return;
} catch (InterruptedException e) {
/* We've been interrupted abnormally */
System.out.println("State History Tree interrupted!"); //$NON-NLS-1$
e.printStackTrace();
} catch (TimeRangeException e) {
/* This also should not happen */
e.printStackTrace();
}
}
}
|
