Commit c5fbd871 authored by Ramon Nou's avatar Ramon Nou
Browse files

Restored FTIO with sizes bug

parent f5e1cba6

src/master.cpp

+46 −32
Original line number Diff line number Diff line
@@ -395,11 +395,16 @@ master_server::transfer_dataset_internal(pending_transfer& pt) { 
        auto fs = FSPlugin::make_fs(

                static_cast<cargo::FSPlugin::type>(s.get_type()));

        struct stat buf;

        fs->stat(p, &buf);

        if(buf.st_mode & S_IFDIR) {

        auto rstat = fs->stat(p, &buf);

        if(rstat == 0 and (buf.st_mode & S_IFDIR)) {

            LOGGER_INFO("Expanding input directory {}", p);

            files = fs->readdir(p);



            // As we need to create a new directory, we need to order the files

            // so that directories are created in the correct order



            // Order the files alphabetically

            std::sort(files.begin(), files.end());

            /*

            We have all the files expanded. Now create a new

            cargo::dataset for each file as s and a new
@@ -430,15 +435,19 @@ master_server::transfer_dataset_internal(pending_transfer& pt) { 


                LOGGER_DEBUG("Expanded file {} -> {}", s_new.path(),

                             d_new.path());

                fs->stat(s_new.path(), &buf);

                rstat = fs->stat(s_new.path(), &buf);

                if(rstat == 0) {

                    if(buf.st_mtime < now) {

                        v_size_new.push_back(buf.st_size);

                        v_s_new.push_back(s_new);

                        v_d_new.push_back(d_new);

                    }

                }

                // break;

            }

        } else {

            fs->stat(s.path(), &buf);

            rstat = fs->stat(s.path(), &buf);

            if(rstat == 0) {

                if(buf.st_mtime < now) {

                    v_s_new.push_back(s);

                    v_d_new.push_back(d);
@@ -446,6 +455,7 @@ master_server::transfer_dataset_internal(pending_transfer& pt) { 
                }

            }

        }

    }



    // empty m_expanded_sources

    pt.m_expanded_sources.assign(v_s_new.begin(), v_s_new.end());
@@ -578,12 +588,14 @@ master_server::transfer_datasets(const network::request& req, 


        } else {

            // We do not create any optimization for single files

            fs->stat(s.path(), &buf);

            rstat = fs->stat(s.path(), &buf);

            if(rstat == 0) {

                v_size_new.push_back(buf.st_size);

                v_s_new.push_back(s);

                v_d_new.push_back(d);

            }

        }

    }



    m_request_manager.create(v_s_new.size(), world.size() - 1)

            .or_else([&](auto&& ec) {
@@ -594,7 +606,8 @@ master_server::transfer_datasets(const network::request& req, 
            .map([&](auto&& r) {

                assert(v_s_new.size() == v_d_new.size());

                if(m_ftio) {

                   // if(sources[0].get_type() == cargo::dataset::type::gekkofs) {

                    // if(sources[0].get_type() ==

                    // cargo::dataset::type::gekkofs) {



                    // We have only one pendingTransfer for FTIO

                    // that can be updated, the issue is that we
@@ -613,11 +626,12 @@ master_server::transfer_datasets(const network::request& req, 


                    // Create the directory if it does not exist (only in

                    // parallel transfer)

                    if(!std::filesystem::path(d.path()).parent_path().empty() and

                    if(!std::filesystem::path(d.path())

                                .parent_path()

                                .empty() and

                       d.supports_parallel_transfer()) {

                        std::filesystem::create_directories(

                                std::filesystem::path(d.path()).parent_path());

                        LOGGER_INFO("Created directory {}", std::filesystem::path(d.path()).parent_path());

                    }

                }