source: trunk/CrypPlugins/KeySearcher/P2P/Storage/StorageHelper.cs @ 2249

Last change on this file since 2249 was 2249, checked in by nolte, 11 years ago

MTC saving...another try

File size: 15.0 KB
Line 
1using System;
2using System.Collections.Generic;
3using System.IO;
4using Cryptool.P2P;
5using Cryptool.P2P.Internal;
6using Cryptool.PluginBase;
7using KeySearcher.P2P.Exceptions;
8using KeySearcher.P2P.Presentation;
9using KeySearcher.P2P.Tree;
10
11namespace KeySearcher.P2P.Storage
12{
13    class StorageHelper
14    {
15        private readonly KeySearcher keySearcher;
16        private readonly StatisticsGenerator statisticsGenerator;
17        private readonly StatusContainer statusContainer;
18
19        //VERSIONNUMBER: Important. Set it +1 manually everytime the length of the MemoryStream Changes
20        private const int version = 4;
21        /*
22        -----------------------------Versionnumber Changelog---------------------------------------------
23        |   Version 1: Added the Versionnumber to the Stream (Leafs)
24        |   Version 2: Added the first User Statistics (Avatar,ID,Count) to the Stream
25        |   Version 3: Added version question (in front of results) + more statistic information (hostname,date) to the Stream
26        |   Version 4: Reorganisation of the DHT structure. Update fully working now + features from previous versions available
27        -------------------------------------------------------------------------------------------------
28         */
29
30        public StorageHelper(KeySearcher keySearcher, StatisticsGenerator statisticsGenerator, StatusContainer statusContainer)
31        {
32            this.keySearcher = keySearcher;
33            this.statisticsGenerator = statisticsGenerator;
34            this.statusContainer = statusContainer;
35        }
36
37        //-------------------------------------------------------------------------------------------
38        //AFTER CHANGING THE FOLLOWING METHODS INCREASE THE VERSION-NUMBER AT THE TOP OF THIS CLASS!
39        //-------------------------------------------------------------------------------------------
40        internal RequestResult UpdateInDht(NodeBase nodeToUpdate)
41        {
42            var memoryStream = new MemoryStream();
43            var binaryWriter = new BinaryWriter(memoryStream);
44
45            //TODO: Append Updater Version
46            binaryWriter.Write('W');
47            binaryWriter.Write(version); 
48
49            if (nodeToUpdate is Node)
50            {
51                UpdateNodeInDht((Node) nodeToUpdate, binaryWriter);
52            } else
53            {
54                UpdateLeafInDht((Leaf) nodeToUpdate, binaryWriter);
55            }
56
57            // Append results
58            binaryWriter.Write(nodeToUpdate.Result.Count);
59            foreach (var valueKey in nodeToUpdate.Result)
60            {
61                binaryWriter.Write(valueKey.key);
62                binaryWriter.Write(valueKey.value);
63                binaryWriter.Write(valueKey.decryption.Length);
64                binaryWriter.Write(valueKey.decryption);
65            }                       
66             
67            //Creating a copy of the activity dictionary
68            var copyAct = nodeToUpdate.Activity;
69
70            //Write number of avatarnames
71            binaryWriter.Write(copyAct.Keys.Count);
72            foreach (string avatar in copyAct.Keys)
73            {
74                var maschCopy = copyAct[avatar];
75                //write avatarname
76                binaryWriter.Write(avatar);
77                //write the number of maschines for this avatar
78                binaryWriter.Write(maschCopy.Keys.Count);
79
80                foreach (long maschID in maschCopy.Keys)
81                {
82                    //write the maschines and their patterncount
83                    binaryWriter.Write(maschID);
84                    binaryWriter.Write(maschCopy[maschID].Count); //int 32
85                    binaryWriter.Write(maschCopy[maschID].Hostname); //String
86                    binaryWriter.Write(maschCopy[maschID].Date.ToBinary()); //DateTime
87                }
88            }
89            return StoreWithStatistic(KeyInDht(nodeToUpdate), memoryStream.ToArray());
90        }
91
92        private static void UpdateNodeInDht(Node nodeToUpdate, BinaryWriter binaryWriter)
93        {
94            binaryWriter.Write(nodeToUpdate.LeftChildFinished);
95            binaryWriter.Write(nodeToUpdate.RightChildFinished);
96        }
97
98        private static void UpdateLeafInDht(Leaf nodeToUpdate, BinaryWriter binaryWriter)
99        {
100            binaryWriter.Write('V');
101            binaryWriter.Write(version);
102            var buffer = nodeToUpdate.LastReservationDate.ToBinary();
103            binaryWriter.Write(buffer);
104            binaryWriter.Write(nodeToUpdate.getClientIdentifier());
105        }
106
107        internal RequestResult UpdateFromDht(NodeBase nodeToUpdate, bool forceUpdate = false)
108        {
109
110            if (!forceUpdate && nodeToUpdate.LastUpdate > DateTime.Now.Subtract(new TimeSpan(0, 0, 5)))
111            {
112                return new RequestResult { Status = RequestResultType.Success };
113            }
114
115            nodeToUpdate.LastUpdate = DateTime.Now;
116
117            var requestResult = RetrieveWithStatistic(KeyInDht(nodeToUpdate));
118            var nodeBytes = requestResult.Data;
119
120            if (nodeBytes == null)
121            {
122                return requestResult;
123            }
124
125            var binaryReader = new BinaryReader(new MemoryStream(nodeBytes));
126
127            //oldVersionFlag will be used to garantee further changes in the Stream
128            var oldVersionFlag = CheckNodeVersion(binaryReader);
129
130            if (nodeToUpdate is Node)
131            {
132                UpdateNodeFromDht((Node) nodeToUpdate, binaryReader);
133            } else
134            {
135                UpdateLeafFromDht((Leaf)nodeToUpdate, binaryReader);
136            }
137
138            // Load results
139            var resultCount = binaryReader.ReadInt32();
140            for (var i = 0; i < resultCount; i++)
141            {
142                var newResult = new KeySearcher.ValueKey
143                                    {
144                                        key = binaryReader.ReadString(),
145                                        value = binaryReader.ReadDouble(),
146                                        decryption = binaryReader.ReadBytes(binaryReader.ReadInt32())
147                                    };
148                nodeToUpdate.Result.AddLast(newResult);
149            }
150           
151            if (binaryReader.BaseStream.Length != binaryReader.BaseStream.Position)
152            { 
153                //Reading the number of avatarnames
154                int avatarcount = binaryReader.ReadInt32();
155                for(int i=0; i<avatarcount;i++)
156                {
157                    //Reading the avatarname and the maschine-count for this name
158                    string avatarname = binaryReader.ReadString();
159                    int maschcount = binaryReader.ReadInt32();
160                    var readMaschcount = new Dictionary<long, Information>();
161                   
162                    /*//previous version 2 didn't had the Information
163                    if (oldVersionFlag < 3)
164                    {
165                        for (int j = 0; j < maschcount; j++)
166                        {
167                            //reading the IDs and patterncount
168                            long maschID = binaryReader.ReadInt64();
169                            int count = binaryReader.ReadInt32();
170                            if (maschID > 0)
171                            {
172                                readMaschcount.Add(maschID, new Information(){Count = count, Hostname = "Unknown", Date = DateTime.MinValue});
173                            }
174                        }
175                    }
176                    else
177                    {
178                     */
179                    for (int j = 0; j < maschcount; j++)
180                    {
181                        //reading the IDs and patterncount
182                        long maschID = binaryReader.ReadInt64();
183                        int count = binaryReader.ReadInt32();
184                        string host = binaryReader.ReadString(); 
185                        var date = DateTime.FromBinary(binaryReader.ReadInt64());
186                        readMaschcount.Add(maschID, new Information() {Count = count, Hostname = host, Date = date});
187                       
188                    }
189                   
190                    if (nodeToUpdate.Activity.ContainsKey(avatarname))
191                    {
192                        nodeToUpdate.Activity[avatarname] = readMaschcount;
193                    }
194                    else
195                    {
196                        nodeToUpdate.Activity.Add(avatarname, readMaschcount);                     
197                    }
198                }               
199            }
200                       
201            if (resultCount > 0)
202            {
203                keySearcher.IntegrateNewResults(nodeToUpdate.Result, nodeToUpdate.Activity, nodeToUpdate.DistributedJobIdentifier);
204                statisticsGenerator.ProcessPatternResults(nodeToUpdate.Result);
205            }
206
207            nodeToUpdate.UpdateCache();
208            return requestResult;
209        }
210
211        private static void UpdateNodeFromDht(Node nodeToUpdate, BinaryReader binaryReader)
212        {
213            nodeToUpdate.LeftChildFinished = binaryReader.ReadBoolean() || nodeToUpdate.LeftChildFinished;
214            nodeToUpdate.RightChildFinished = binaryReader.ReadBoolean() || nodeToUpdate.RightChildFinished;
215        }
216
217        private static void UpdateLeafFromDht(Leaf nodeToUpdate, BinaryReader binaryReader)
218        {
219            var oldVersionFlag = CheckVersion(binaryReader);
220               
221            var date = DateTime.FromBinary(binaryReader.ReadInt64());
222            if (date > nodeToUpdate.LastReservationDate)
223            {
224                nodeToUpdate.LastReservationDate = date;
225            }
226           
227            try
228            {
229                if (binaryReader.BaseStream.Length - binaryReader.BaseStream.Position >= 8)
230                {
231                    nodeToUpdate.setClientIdentifier(binaryReader.ReadInt64());
232                }
233                else
234                {
235                    throw new Exception();
236                }
237            }
238            catch (Exception)
239            {
240                // client id not available, use default
241                nodeToUpdate.setClientIdentifier(-1);
242            }
243           
244        }
245
246        internal static string KeyInDht(NodeBase node)
247        {
248            return string.Format("{0}_node_{1}_{2}", node.DistributedJobIdentifier, node.From, node.To);
249        }
250
251        private static int CheckVersion(BinaryReader binaryReader)
252        {           
253            try
254            {
255                //Checking if there's a version in the stream
256                int vers = binaryReader.PeekChar();
257                if (vers == 86) //V infront of a Leaf
258                {
259                    //Reading the char and the versionnumber
260                    char magic = binaryReader.ReadChar();
261                    int versionInUse = binaryReader.ReadInt32();
262                    //Check if a newer Version is in use
263                    if (versionInUse > version)
264                    {
265                        throw new KeySearcherStopException();
266                    }
267                    return versionInUse;
268                }
269                else
270                {
271                    return 0;
272                }
273            }
274            catch(KeySearcherStopException)
275            {
276                throw new KeySearcherStopException();
277            }           
278        }
279
280        private static int CheckNodeVersion(BinaryReader binaryReader)
281        {
282            try
283            {
284                //Checking if there's a version in the stream
285                int vers = binaryReader.PeekChar();
286                if (vers == 87) //W infront of a Node
287                {
288                    //Reading the char and the versionnumber
289                    char magic = binaryReader.ReadChar();
290                    int versionInUse = binaryReader.ReadInt32();
291                    //Check if a newer Version is in use
292                    if (versionInUse > version)
293                    {
294                        throw new KeySearcherStopException();
295                    }
296                    return versionInUse;
297                }
298                else
299                {
300                    return 0;
301                }
302            }
303            catch (KeySearcherStopException)
304            {
305                throw new KeySearcherStopException();
306            }
307        }
308
309        public DateTime StartDate(String ofJobIdentifier)
310        {
311            var key = ofJobIdentifier + "_startdate";
312            var requestResult = RetrieveWithStatistic(key);
313
314            if (requestResult.IsSuccessful() && requestResult.Data != null)
315            {
316                var startTimeUtc = DateTime.SpecifyKind(
317                    DateTime.FromBinary(BitConverter.ToInt64(requestResult.Data, 0)), DateTimeKind.Utc);
318                return startTimeUtc.ToLocalTime();
319            }
320
321            StoreWithStatistic(key, BitConverter.GetBytes((DateTime.UtcNow.ToBinary())));
322            return DateTime.Now;
323        }
324
325        public long SubmitterID(String ofJobIdentifier)
326        {
327            var key = ofJobIdentifier + "_submitterid";
328            var requestResult = RetrieveWithStatistic(key);
329
330            if (requestResult.IsSuccessful() && requestResult.Data != null)
331            {
332                var submitterid = BitConverter.ToInt64(requestResult.Data, 0);
333                return submitterid;
334            }
335
336            StoreWithStatistic(key, BitConverter.GetBytes(Cryptool.PluginBase.Miscellaneous.UniqueIdentifier.GetID()));
337            return Cryptool.PluginBase.Miscellaneous.UniqueIdentifier.GetID();
338        }
339
340        public RequestResult RetrieveWithStatistic(string key)
341        {
342            statusContainer.RetrieveRequests++;
343            statusContainer.TotalDhtRequests++;
344            var requestResult = P2PManager.Retrieve(key);
345
346            if (requestResult.Data != null)
347            {
348                statusContainer.RetrievedBytes += requestResult.Data.Length;
349                statusContainer.TotalBytes += requestResult.Data.Length;
350            }
351
352            return requestResult;
353        }
354
355        public RequestResult RemoveWithStatistic(string key)
356        {
357            statusContainer.RemoveRequests++;
358            statusContainer.TotalDhtRequests++;
359            return P2PManager.Remove(key);
360        }
361
362        public RequestResult StoreWithStatistic(string key, byte[] data)
363        {
364            statusContainer.StoreRequests++;
365            statusContainer.TotalDhtRequests++;
366            var requestResult = P2PManager.Store(key, data);
367
368            if (requestResult.Data != null)
369            {
370                statusContainer.StoredBytes += requestResult.Data.Length;
371                statusContainer.TotalBytes += requestResult.Data.Length;
372            }
373
374            return requestResult;
375        }
376    }
377}
Note: See TracBrowser for help on using the repository browser.