Project

General

Profile

Statistics
| Branch: | Revision:

gdp-if / tensorflow / gdpfs_tf.cc @ master

History | View | Annotate | Download (13.7 KB)

1
#include "gdpfs_tf.h"
2

    
3
#include <unistd.h>
4
#include <stdio.h>
5
#include <errno.h>
6
#include <string.h>
7
#include <sys/stat.h>
8
#include <cstdlib>
9
#include <string>
10
#include <iostream>
11
#include <map>
12

    
13
#include "tensorflow/core/lib/core/error_codes.pb.h"
14
#include "tensorflow/core/lib/core/status.h"
15
#include "tensorflow/core/lib/strings/strcat.h"
16
#include "tensorflow/core/platform/env.h"
17
#include "tensorflow/core/platform/file_system_helper.h"
18
#include "tensorflow/core/platform/logging.h"
19

    
20
#include "gdpfs.h"
21

    
22
namespace tensorflow {
23

    
24
static EP_DBG Dbg = EP_DBG_INIT("gdpfs.tf",
25
                                                "GDP FileSystem: Tensorflow Extensions");
26

    
27

    
28
/*
29
**        First, we implement a few classes that conform to
30
**  the behavior expected by the filesystem users.
31
**
32
**  These classes are thin wrappers around the classes
33
**  provided by the gdpfs. Note that the WritableFile
34
**  and AppendableFile are implemented by the same class.
35
*/
36

    
37
/*
38
** RandomAccessFile implementation based on GDPFileRO
39
** (provided by gdpfs).
40
*/
41
class GDPRandomAccessFile : public RandomAccessFile
42
{
43

    
44
        public:
45
        GDPRandomAccessFile(const string& fname)
46
        {
47
                ep_dbg_cprintf(Dbg, 4, "New RandomAccessFile(%s)\n",
48
                                                                                        fname.c_str());
49
                _myname = fname;
50
                _file = new GDPFileRO(fname);
51
        }
52

    
53
        ~GDPRandomAccessFile() override { delete _file; }
54

    
55
        Status Read(uint64 offset, size_t n, StringPiece* result,
56
                                                        char* scratch) const override
57
        {
58
                ep_dbg_cprintf(Dbg, 24, "Reading [%lld +(%ld)] bytes\n", offset, n);
59
                size_t read = _file->Read(offset, n, scratch);
60
                *result = StringPiece(scratch, read);
61
                if (read == n)
62
                        return Status::OK();
63
                else
64
                {
65
                        ep_dbg_cprintf(Dbg, 11,
66
                                                        "Read only %ld bytes (asked for %ld)\n", read, n);
67
                        return Status(error::OUT_OF_RANGE,
68
                                                        "Read less bytes than requested");
69
                }
70
        }
71

    
72
        private:
73
        GDPFileRO* _file;
74
        string _myname;
75
};
76

    
77

    
78
/*
79
** WritableFile implementation based on GDPFileWO
80
** (provided by gdpfs).
81
*/
82

    
83
class GDPWritableFile : public WritableFile
84
{
85

    
86
        public:
87
        GDPWritableFile(const string& fname)
88
        {
89
                ep_dbg_cprintf(Dbg, 4, "New WritableFile(%s)\n",
90
                                                                                        fname.c_str());
91
                _myname = fname;
92
                _file = new GDPFileWO(fname);
93
        }
94

    
95
        ~GDPWritableFile() override { delete _file; }
96

    
97
        Status Append(const StringPiece& data) override
98
        {
99
                string s(data.data(), data.size());
100
                _file->Append(s);
101
                return Status::OK();
102
        }
103

    
104
        Status Close() override
105
        { 
106
                ep_dbg_cprintf(Dbg, 6, "Close(%s)\n", _myname.c_str());
107
                return Status::OK();
108
        }
109
        Status Flush() override
110
        {
111
                ep_dbg_cprintf(Dbg, 6, "Flush(%s)\n", _myname.c_str());
112
                return Status::OK();
113
        }
114
        Status Sync() override
115
        {
116
                ep_dbg_cprintf(Dbg, 6, "Sync(%s)\n", _myname.c_str());
117
                return Status::OK();
118
        }
119

    
120
        private:
121
        GDPFileWO* _file;
122
        string _myname;
123

    
124
};
125

    
126
/*
127
** MemoryMapped implementation based on GDPFileROMemMap
128
** (provided by gdpfs).
129
*/
130
class GDPReadOnlyMemoryRegion : public ReadOnlyMemoryRegion
131
{
132
        public:
133
        GDPReadOnlyMemoryRegion(const string& fname)
134
        {
135
                ep_dbg_cprintf(Dbg, 4, "New ReadOnlyMemoryRegion(%s)\n",
136
                                                                                        fname.c_str());
137
                _myname = fname;
138
                _file = new GDPFileROMemMap(fname);
139
        }
140

    
141
        ~GDPReadOnlyMemoryRegion() override { delete _file; }
142

    
143
        const void* data() override
144
        {
145
                ep_dbg_cprintf(Dbg, 9, "%s.data()\n", _myname.c_str());
146
                return (const char*) _file->data();
147
        }
148

    
149
        uint64 length() override
150
        {
151
                ep_dbg_cprintf(Dbg, 9, "%s.length()\n", _myname.c_str());
152
                return (uint64) _file->length();
153
        }
154

    
155
        private:
156
        GDPFileROMemMap* _file;
157
        string _myname;
158

    
159
};
160

    
161
/**********************************************************/
162
/******** Implementation of GDPFileSystem below ***********/
163
/**********************************************************/
164

    
165

    
166
GDPFileSystem::GDPFileSystem()
167
{
168
        // nothing to see here.
169
        GDPfsInit(GDP_MODE_RA);
170
        ep_dbg_cprintf(Dbg, 2, "Initializing file system\n");
171
}
172

    
173
GDPFileSystem::~GDPFileSystem()
174
{
175
        ep_dbg_cprintf(Dbg, 2, "Freeing file system\n");
176

    
177
        // we close the directories in _open_root_dirs
178
        std::map<string, GDPDir*>::iterator it;
179
        for (it=_open_root_dirs.begin(); it!=_open_root_dirs.end(); it++)
180
        {
181
                ep_dbg_cprintf(Dbg, 15, "Freeing up open dir %s\n",
182
                                                                                        it->first.c_str());
183
                delete it->second;
184
        }
185
}
186

    
187
GDPDir* GDPFileSystem::get_root_dir(const string& dirname)
188
{
189

    
190
        ep_dbg_cprintf(Dbg, 14, "get_root_dir(%s)\n", dirname.c_str());
191

    
192
        std::map<string, GDPDir*>::iterator it;
193
        it = _open_root_dirs.find(dirname);
194
        if (it == _open_root_dirs.end())
195
        {
196
                // open the directory, and return the pointer
197
                // dirname is also the logname for the topmost
198
                // directory; this is intentional.
199
                ep_dbg_cprintf(Dbg, 14, "Opening log/dir %s\n", dirname.c_str());
200
                GDPDir *ptr = new GDPDir(dirname, GDP_MODE_RA);
201
                _open_root_dirs[dirname] = ptr;
202
                return ptr;
203
        }
204
        else
205
        {
206
                ep_dbg_cprintf(Dbg, 14, "Returning alredy open log/dir %s\n",
207
                                                                                                        dirname.c_str());
208
                return it->second;
209
        }
210
}
211

    
212
Status GDPFileSystem::NewRandomAccessFile(
213
                const string& fname,
214
                std::unique_ptr<RandomAccessFile>*result)
215
{
216
        ep_dbg_cprintf(Dbg, 4, "NewRandomAccessFile(%s)\n", fname.c_str());
217

    
218
        string rootlog, path, logname;
219
        GDPDir *ptr;
220
        assert (ParsePath(fname, &rootlog, &path) == kSuccess);
221
        ptr = get_root_dir(rootlog);
222

    
223
        if (ptr->NameExists(path))
224
        {
225
                ptr->GetEntryLogname(path, &logname);
226

    
227
                ep_dbg_cprintf(Dbg, 7, "[%s]/[%s] => [%s]\n",
228
                                                                rootlog.c_str(), path.c_str(),
229
                                                                logname.c_str());
230

    
231
                result->reset(new GDPRandomAccessFile(logname));
232
                return Status::OK();
233
        }
234
        else
235
        {
236
                ep_dbg_cprintf(Dbg, 7, "[%s]/[%s] Does NOT Exist\n",
237
                                                                rootlog.c_str(), path.c_str());
238
                result->reset(NULL);
239
                return errors::NotFound("File doesn't exist\n");
240
        }
241
}
242

    
243
Status GDPFileSystem::NewReadOnlyMemoryRegionFromFile(
244
                                const string& fname,
245
                                std::unique_ptr<ReadOnlyMemoryRegion>* result)
246
{
247
        ep_dbg_cprintf(Dbg, 4, "NewReadOnlyMemoryRegionFromFile(%s)\n",
248
                                                                                                        fname.c_str());
249
        GDPDir *ptr;
250
        string rootlog, path, logname;
251
        assert (ParsePath(fname, &rootlog, &path) == kSuccess);
252
        ptr = get_root_dir(rootlog);
253

    
254
        if (ptr->NameExists(path))
255
        {
256
                ptr->GetEntryLogname(path, &logname);
257

    
258
                ep_dbg_cprintf(Dbg, 7, "[%s]/[%s] => [%s]\n",
259
                                                                rootlog.c_str(), path.c_str(),
260
                                                                logname.c_str());
261

    
262
                result->reset(new GDPReadOnlyMemoryRegion(logname));
263
                return Status::OK();
264
        }
265
        else
266
        {
267
                ep_dbg_cprintf(Dbg, 7, "[%s]/[%s] Does NOT Exist\n",
268
                                                                rootlog.c_str(), path.c_str());
269
                result->reset(NULL);
270
                return errors::NotFound("File doesn't exist\n");
271
        }
272
}
273

    
274
Status GDPFileSystem::NewWritableFile(
275
                const string& fname,
276
                std::unique_ptr<WritableFile>* result)
277
{
278
        ep_dbg_cprintf(Dbg, 4, "NewWritableFile(%s)\n", fname.c_str());
279

    
280
        GDPDir *ptr;
281
        string rootlog, path, logname;
282
        assert (ParsePath(fname, &rootlog, &path) == kSuccess);
283
        ptr = get_root_dir(rootlog);
284

    
285
        if (ptr->NameExists(path))
286
        {
287
                ep_dbg_cprintf(Dbg, 7, "Opening existing log as writable file\n");
288
                ptr->GetEntryLogname(path, &logname);
289
        }
290
        else
291
        {
292
                ep_dbg_cprintf(Dbg, 7, "Creating new log as writable file\n");
293
                ptr->NewFile(path, &logname);
294
        }
295

    
296
        result->reset(new GDPWritableFile(logname));
297
        ep_dbg_cprintf(Dbg, 7, "[%s]/[%s] => [%s]\n",
298
                                                                rootlog.c_str(), path.c_str(),
299
                                                                logname.c_str());
300
        return Status::OK();
301
}
302

    
303
Status GDPFileSystem::NewAppendableFile(
304
                                const string& fname,
305
                                std::unique_ptr<WritableFile>* result)
306
{
307
        ep_dbg_cprintf(Dbg, 4, "NewAppendableFile(%s)\n", fname.c_str());
308
        return NewWritableFile(fname, result);
309
}
310

    
311

    
312
Status GDPFileSystem::FileExists(const string& fname)
313
{
314
        ep_dbg_cprintf(Dbg, 4, "FileExists(%s)\n", fname.c_str());
315
        string rootlog, path;
316
        GDPDir *ptr;
317

    
318
        assert (ParsePath(fname, &rootlog, &path) == kSuccess);
319
        ptr = get_root_dir(rootlog);
320

    
321
        if (ptr->NameExists(path))
322
        {
323
                ep_dbg_cprintf(Dbg, 7, "[%s]/[%s] Exists\n",
324
                                                                rootlog.c_str(), path.c_str());
325
                return Status::OK();
326
        }
327
        else
328
        {
329
                ep_dbg_cprintf(Dbg, 7, "[%s]/[%s] Does NOT Exist\n",
330
                                                                rootlog.c_str(), path.c_str());
331
                return errors::NotFound("Not found\n");
332
        }
333
}
334

    
335
Status GDPFileSystem::GetChildren(
336
                                const string& dir,
337
                                std::vector<string>* result)
338
{
339
        ep_dbg_cprintf(Dbg, 4, "GetChildren(%s)\n", dir.c_str());
340

    
341
        string rootlog, path;
342
        GDPDir *ptr;
343
        std::vector<string> tmp;
344
        std::vector<string>::iterator it;
345
        string s;
346

    
347
        assert (ParsePath(dir, &rootlog, &path) == kSuccess);
348
        ptr = get_root_dir(rootlog);
349

    
350
        ptr->GetChildren(path, &tmp);
351

    
352
        // Make sure we translate the names back to full path
353
        for (it=tmp.begin(); it!=tmp.end(); it++)
354
        {
355
                s.assign("gdp://");
356
                s.append(rootlog);
357
                s.append("/");
358
                s.append(*it);
359
                result->push_back(s);
360
        }
361

    
362
        return Status::OK();
363

    
364
        /*
365
        // XXX This should be removed after testing GDPDir's GetChildren
366

367
        for (it=ptr->dentries_.begin(); it!= ptr->dentries_.end(); it++)
368
        {
369
                BaseDirName(it->first, &dirname, &basename);
370
                fulldirname.assign("gdp://");
371
                fulldirname.append(rootlog);
372
                fulldirname.append("/");
373
                fulldirname.append(dirname);
374

375
                ep_dbg_cprintf(Dbg, 15, "comparing %s with %s\n",
376
                                                                fulldirname.c_str(), dir.c_str());
377
                if (fulldirname.compare(dir) == 0)
378
                {
379
                        ep_dbg_cprintf(Dbg, 12, "child => %s\n", basename.c_str());
380
                        result->push_back(basename);
381
                }
382
        }
383

384
        return Status::OK();
385
        */
386
}
387

    
388

    
389
Status GDPFileSystem::Stat(const string& fname, FileStatistics* stat)
390
{
391
        ep_dbg_cprintf(Dbg, 4, "Stat(%s)\n", fname.c_str());
392
        string rootlog, path;
393
        GDPDir *ptr;
394
        GdpStat tmp;
395

    
396
        assert (ParsePath(fname, &rootlog, &path) == kSuccess);
397
        ptr = get_root_dir(rootlog);
398

    
399
        ptr->Stat(path, &tmp);
400

    
401
        stat->is_directory = tmp.is_directory;
402
        stat->length = tmp.length;
403
        stat->mtime_nsec = tmp.mtime_nsec;
404
        
405
        /*
406

407
        stat->is_directory = (ptr->GetEntryType(path) == GDPfs::DIR);
408
        if (!stat->is_directory)
409
        {
410
                GetFileSize(fname, &size);
411
                stat->length = size;
412

413
                // getting last update time
414
                ptr->GetEntryLogname(path, &logname);
415
                GDPFileLowLevel f(logname, GDP_MODE_RO);
416
                stat->mtime_nsec = f.GetMTime();
417
        }
418
        else
419
        {
420
                stat->length = 0;
421
                stat->mtime_nsec = TimeNS();        // XXX: fix this
422
        }
423

424
        */
425
        ep_dbg_cprintf(Dbg, 4, "Stat(%s)=%lld, %lld\n", fname.c_str(),
426
                                                                        stat->length, stat->mtime_nsec);
427
        return Status::OK();
428
}
429

    
430
Status GDPFileSystem::GetMatchingPaths(
431
                                const string& pattern,
432
                                std::vector<string>* results)
433
{
434
        ep_dbg_cprintf(Dbg, 4, "GetMatchingPaths(%s)\n", pattern.c_str());
435
        ep_dbg_cprintf(Dbg, 1, "Note that this isn't fully implemented yet.\n");
436

    
437
        string rootlog, path;
438
        GDPDir *ptr;
439
        std::vector<string> tmp;
440
        std::vector<string>::iterator it;
441
        string s;
442

    
443
        assert (ParsePath(pattern, &rootlog, &path) == kSuccess);
444
        ptr = get_root_dir(rootlog);
445

    
446
        ptr->GetMatchingPaths(path, &tmp);
447

    
448
        // Make sure we translate the names back to full path
449
        for (it=tmp.begin(); it!= tmp.end(); it++)
450
        {
451
                s.assign("gdp://");
452
                s.append(rootlog);
453
                s.append("/");
454
                s.append(*it);
455
                results->push_back(s);
456
        }
457

    
458
        return Status::OK();
459

    
460
        /*
461

462
        // XXX: this should be removed once GDPDir GetMatchingPaths is tested.
463

464
        if (pattern.find('*')!=string::npos)
465
                return errors::Unimplemented(" '*' Not yet implemented\n");
466
        if (pattern.find('?')!=string::npos)
467
                return errors::Unimplemented(" '?' Not yet implemented\n");
468
        if (pattern.find('[')!=string::npos)
469
                return errors::Unimplemented(" '[' Not yet implemented\n");
470
        if (pattern.find(']')!=string::npos)
471
                return errors::Unimplemented(" ']' Not yet implemented\n");
472
        if (pattern.find("\\\\")!=string::npos)
473
                return errors::Unimplemented(" ds Not yet implemented\n");
474
        
475
        size_t pos;
476
        string rootlog, path;
477
        GDPDir *ptr;
478
        std::map<string, string>::iterator it;
479

480
        assert (ParsePath(pattern, &rootlog, &path) == kSuccess);
481
        ptr = get_root_dir(rootlog);
482

483
        for (it=ptr->dentries_.begin(); it!=ptr->dentries_.end(); it++)
484
        {
485
                ep_dbg_cprintf(Dbg, 30, "Matching %s against: %s\n",
486
                                                                        path.c_str(), it->first.c_str());
487
                pos = it->first.find(path);
488
                if (pos == 0)
489
                {
490
                        ep_dbg_cprintf(Dbg, 22, "Found match %s\n", it->first.c_str());
491
                        results->push_back(it->first);
492
                }
493
        }
494

495
        return Status::OK();
496
        */
497
}
498

    
499
Status GDPFileSystem::DeleteFile(const string& fname)
500
{
501
        ep_dbg_cprintf(Dbg, 4, "DeleteFile(%s)\n", fname.c_str());
502
        string rootlog, path;
503
        GDPDir *ptr;
504

    
505
        assert (ParsePath(fname, &rootlog, &path) == kSuccess);
506
        ptr = get_root_dir(rootlog);
507

    
508
        ep_dbg_cprintf(Dbg, 7, "Deleting file [%s]/[%s]\n",
509
                                                        rootlog.c_str(), path.c_str());
510

    
511
        ptr->DeleteFile(path);
512
        return Status::OK();
513
}
514

    
515
Status GDPFileSystem::CreateDir(const string& name)
516
{
517
        ep_dbg_cprintf(Dbg, 4, "CreateDir(%s)\n", name.c_str());
518
        string rootlog, path;
519
        GDPDir *ptr;
520

    
521
        assert (ParsePath(name, &rootlog, &path) == kSuccess);
522
        ptr = get_root_dir(rootlog);
523

    
524
        ptr->CreateDir(path);
525
        ep_dbg_cprintf(Dbg, 7, "New dir [%s]/[%s] => [X]\n",
526
                                                        rootlog.c_str(), path.c_str());
527

    
528
        return Status::OK();
529
}
530

    
531

    
532
Status GDPFileSystem::DeleteDir(const string& name)
533
{
534
        ep_dbg_cprintf(Dbg, 4, "DeleteDir(%s)\n", name.c_str());
535
        string rootlog, path;
536
        GDPDir *ptr;
537

    
538
        assert (ParsePath(name, &rootlog, &path) == kSuccess);
539
        ptr = get_root_dir(rootlog);
540

    
541
        ep_dbg_cprintf(Dbg, 7, "Deleting dir [%s]/[%s]\n",
542
                                                        rootlog.c_str(), path.c_str());
543

    
544
        ptr->DeleteDir(path);
545
        return Status::OK();
546
}
547

    
548

    
549
Status GDPFileSystem::GetFileSize(const string& fname, uint64* size)
550
{
551
        ep_dbg_cprintf(Dbg, 4, "GetFileSize(%s)\n", fname.c_str());
552
        string rootlog, path, logname;
553
        GDPDir *ptr;
554

    
555
        assert (ParsePath(fname, &rootlog, &path) == kSuccess);
556
        ptr = get_root_dir(rootlog);
557
        assert (ptr->GetEntryType(path) == GDPfs::FILE);
558

    
559
        ptr->GetEntryLogname(path, &logname);
560
        GDPFile f(logname, GDP_MODE_RO);
561
        *size = (uint64) f.GetFileSize();
562

    
563
        return Status::OK();
564
}
565

    
566

    
567
Status GDPFileSystem::RenameFile(const string& src, const string& target)
568
{
569
        ep_dbg_cprintf(Dbg, 4, "RenameFile(%s=>%s)\n",
570
                                                                src.c_str(), target.c_str());
571

    
572
        string rootlog1, path1, rootlog2, path2;
573
        GDPDir *ptr;
574

    
575
        assert (ParsePath(src, &rootlog1, &path1) == kSuccess);
576
        assert (ParsePath(target, &rootlog2, &path2) == kSuccess);
577

    
578
        // we don't support copying across different root
579
        // directories.
580
        assert (rootlog1.compare(rootlog2) == 0);
581

    
582
        ptr = get_root_dir(rootlog1);
583

    
584
        ptr->RenameFile(path1, path2);
585
        return Status::OK();
586
}
587

    
588
REGISTER_FILE_SYSTEM("gdp", GDPFileSystem);
589

    
590
}        // namespace tensorflow