1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.giraph.ooc.command;
20
21 import org.apache.giraph.ooc.OutOfCoreEngine;
22 import org.apache.giraph.ooc.data.DiskBackedEdgeStore;
23 import org.apache.giraph.ooc.data.DiskBackedMessageStore;
24 import org.apache.giraph.ooc.data.DiskBackedPartitionStore;
25
26 import java.io.IOException;
27
28
29
30
31 public class StoreDataBufferIOCommand extends IOCommand {
32
33
34
35
36 public enum DataBufferType { PARTITION, MESSAGE };
37
38
39
40 private final DataBufferType type;
41
42
43
44
45
46
47
48
49 public StoreDataBufferIOCommand(OutOfCoreEngine oocEngine,
50 int partitionId,
51 DataBufferType type) {
52 super(oocEngine, partitionId);
53 this.type = type;
54 }
55
56 @Override
57 public boolean execute() throws IOException {
58 boolean executed = false;
59 if (oocEngine.getMetaPartitionManager()
60 .startOffloadingBuffer(partitionId)) {
61 switch (type) {
62 case PARTITION:
63 DiskBackedPartitionStore partitionStore =
64 (DiskBackedPartitionStore)
65 oocEngine.getServerData().getPartitionStore();
66 numBytesTransferred +=
67 partitionStore.offloadBuffers(partitionId);
68 DiskBackedEdgeStore edgeStore =
69 (DiskBackedEdgeStore) oocEngine.getServerData().getEdgeStore();
70 numBytesTransferred += edgeStore.offloadBuffers(partitionId);
71 break;
72 case MESSAGE:
73 DiskBackedMessageStore messageStore =
74 (DiskBackedMessageStore)
75 oocEngine.getServerData().getIncomingMessageStore();
76 numBytesTransferred +=
77 messageStore.offloadBuffers(partitionId);
78 break;
79 default:
80 throw new IllegalStateException("execute: requested data buffer type " +
81 "does not exist!");
82 }
83 oocEngine.getMetaPartitionManager().doneOffloadingBuffer(partitionId);
84 executed = true;
85 }
86 return executed;
87 }
88
89 @Override
90 public IOCommandType getType() {
91 return IOCommandType.STORE_BUFFER;
92 }
93
94 @Override
95 public String toString() {
96 return "StoreDataBufferIOCommand: (partitionId = " + partitionId + ", " +
97 "type = " + type.name() + ")";
98 }
99 }