examples/pipeline/square/square.cpp

   1 /*
   2     Copyright (c) 2005-2019 Intel Corporation
   3
   4     Licensed under the Apache License, Version 2.0 (the "License");
   5     you may not use this file except in compliance with the License.
   6     You may obtain a copy of the License at
   7
   8         http://www.apache.org/licenses/LICENSE-2.0
   9
  10     Unless required by applicable law or agreed to in writing, software
  11     distributed under the License is distributed on an "AS IS" BASIS,
  12     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13     See the License for the specific language governing permissions and
  14     limitations under the License.
  15 */
  16
  17 //
  18 // Example program that reads a file of decimal integers in text format
  19 // and changes each to its square.
  20 //
  21 #include "tbb/pipeline.h"
  22 #include "tbb/tick_count.h"
  23 #include "tbb/tbb_allocator.h"
  24 #include <cstring>
  25 #include <cstdlib>
  26 #include <cstdio>
  27 #include <cctype>
  28 #include "tbb/global_control.h"
  29 #include "../../common/utility/utility.h"
  30 #include "../../common/utility/get_default_num_threads.h"
  31
  32 extern void generate_if_needed(const char*);
  33
  34 using namespace std;
  35
  36 //! Holds a slice of text.
  37 /** Instances *must* be allocated/freed using methods herein, because the C++ declaration
  38     represents only the header of a much larger object in memory. */
  39 class TextSlice {
  40     //! Pointer to one past last character in sequence
  41     char* logical_end;
  42     //! Pointer to one past last available byte in sequence.
  43     char* physical_end;
  44 public:
  45     //! Allocate a TextSlice object that can hold up to max_size characters.
  46     static TextSlice* allocate( size_t max_size ) {
  47         // +1 leaves room for a terminating null character.
  48         TextSlice* t = (TextSlice*)tbb::tbb_allocator<char>().allocate( sizeof(TextSlice)+max_size+1 );
  49         t->logical_end = t->begin();
  50         t->physical_end = t->begin()+max_size;
  51         return t;
  52     }
  53     //! Free a TextSlice object
  54     void free() {
  55         tbb::tbb_allocator<char>().deallocate((char*)this,sizeof(TextSlice)+(physical_end-begin())+1);
  56     }
  57     //! Pointer to beginning of sequence
  58     char* begin() {return (char*)(this+1);}
  59     //! Pointer to one past last character in sequence
  60     char* end() {return logical_end;}
  61     //! Length of sequence
  62     size_t size() const {return logical_end-(char*)(this+1);}
  63     //! Maximum number of characters that can be appended to sequence
  64     size_t avail() const {return physical_end-logical_end;}
  65     //! Append sequence [first,last) to this sequence.
  66     void append( char* first, char* last ) {
  67         memcpy( logical_end, first, last-first );
  68         logical_end += last-first;
  69     }
  70     //! Set end() to given value.
  71     void set_end( char* p ) {logical_end=p;}
  72 };
  73
  74 size_t MAX_CHAR_PER_INPUT_SLICE = 4000;
  75 string InputFileName = "input.txt";
  76 string OutputFileName = "output.txt";
  77
  78 TextSlice* next_slice = NULL;
  79
  80 class MyInputFunc {
  81 public:
  82     MyInputFunc(FILE* input_file_);
  83     MyInputFunc(const MyInputFunc& f) : input_file(f.input_file) { }
  84     ~MyInputFunc();
  85     TextSlice* operator()(tbb::flow_control& fc) const;
  86 private:
  87     FILE* input_file;
  88 };
  89
  90 MyInputFunc::MyInputFunc(FILE* input_file_) :
  91     input_file(input_file_) { }
  92
  93 MyInputFunc::~MyInputFunc() {
  94 }
  95
  96 TextSlice* MyInputFunc::operator()(tbb::flow_control& fc) const {
  97     // Read characters into space that is available in the next slice.
  98     if (!next_slice)
  99         next_slice = TextSlice::allocate(MAX_CHAR_PER_INPUT_SLICE);
 100     size_t m = next_slice->avail();
 101     size_t n = fread(next_slice->end(), 1, m, input_file);
 102     if (!n && next_slice->size() == 0) {
 103         // No more characters to process
 104         fc.stop();
 105         return NULL;
 106     }
 107     else {
 108         // Have more characters to process.
 109         TextSlice* t = next_slice;
 110         next_slice = TextSlice::allocate(MAX_CHAR_PER_INPUT_SLICE);
 111         char* p = t->end() + n;
 112         if (n == m) {
 113             // Might have read partial number.
 114             // If so, transfer characters of partial number to next slice.
 115             while (p > t->begin() && isdigit(p[-1]))
 116                 --p;
 117             assert(p > t->begin()); // Number too large to fit in buffer
 118             next_slice->append(p, t->end() + n);
 119         }
 120         t->set_end(p);
 121         return t;
 122     }
 123 }
 124
 125 // Functor that changes each decimal number to its square.
 126 class MyTransformFunc {
 127 public:
 128     TextSlice* operator()(TextSlice* input) const;
 129 };
 130
 131 TextSlice* MyTransformFunc::operator()(TextSlice* input) const {
 132     // Add terminating null so that strtol works right even if number is at end of the input.
 133     *input->end() = '\0';
 134     char* p = input->begin();
 135     TextSlice* out = TextSlice::allocate(2 * MAX_CHAR_PER_INPUT_SLICE);
 136     char* q = out->begin();
 137     for (;;) {
 138         while (p < input->end() && !isdigit(*p))
 139             *q++ = *p++;
 140         if (p == input->end())
 141             break;
 142         long x = strtol(p, &p, 10);
 143         // Note: no overflow checking is needed here, as we have twice the
 144         // input string length, but the square of a non-negative integer n
 145         // cannot have more than twice as many digits as n.
 146         long y = x * x;
 147         sprintf(q, "%ld", y);
 148         q = strchr(q, 0);
 149     }
 150     out->set_end(q);
 151     input->free();
 152     return out;
 153 }
 154
 155 // Functor that writes a TextSlice to a file.
 156 class MyOutputFunc {
 157     FILE* my_output_file;
 158 public:
 159     MyOutputFunc(FILE* output_file);
 160     void operator()(TextSlice* item) const;
 161 };
 162
 163 MyOutputFunc::MyOutputFunc(FILE* output_file) :
 164     my_output_file(output_file)
 165 {
 166 }
 167
 168 void MyOutputFunc::operator()(TextSlice* out) const {
 169     size_t n = fwrite(out->begin(), 1, out->size(), my_output_file);
 170     if (n != out->size()) {
 171         fprintf(stderr, "Can't write into file '%s'\n", OutputFileName.c_str());
 172         exit(1);
 173     }
 174     out->free();
 175 }
 176
 177 bool silent = false;
 178
 179 int run_pipeline( int nthreads )
 180 {
 181     FILE* input_file = fopen( InputFileName.c_str(), "r" );
 182     if( !input_file ) {
 183         throw std::invalid_argument( ("Invalid input file name: "+InputFileName).c_str() );
 184         return 0;
 185     }
 186     FILE* output_file = fopen( OutputFileName.c_str(), "w" );
 187     if( !output_file ) {
 188         throw std::invalid_argument( ("Invalid output file name: "+OutputFileName).c_str() );
 189         return 0;
 190     }
 191
 192     tbb::tick_count t0 = tbb::tick_count::now();
 193
 194     // Need more than one token in flight per thread to keep all threads
 195     // busy; 2-4 works
 196     tbb::parallel_pipeline(
 197         nthreads*4,
 198         tbb::make_filter<void,TextSlice*>(
 199             tbb::filter::serial_in_order, MyInputFunc(input_file) )
 200     &
 201         tbb::make_filter<TextSlice*,TextSlice*>(
 202             tbb::filter::parallel, MyTransformFunc() )
 203     &
 204         tbb::make_filter<TextSlice*,void>(
 205             tbb::filter::serial_in_order, MyOutputFunc(output_file) )
 206     );
 207
 208     tbb::tick_count t1 = tbb::tick_count::now();
 209
 210     fclose( output_file );
 211     fclose( input_file );
 212
 213     if ( !silent ) printf("time = %g\n", (t1-t0).seconds());
 214
 215     return 1;
 216 }
 217
 218 int main( int argc, char* argv[] ) {
 219     try {
 220         tbb::tick_count mainStartTime = tbb::tick_count::now();
 221
 222         // The 1st argument is the function to obtain 'auto' value; the 2nd is the default value
 223         // The example interprets 0 threads as "run serially, then fully subscribed"
 224         utility::thread_number_range threads( utility::get_default_num_threads, 0 );
 225
 226         utility::parse_cli_arguments(argc,argv,
 227             utility::cli_argument_pack()
 228             //"-h" option for displaying help is present implicitly
 229             .positional_arg(threads,"n-of-threads",utility::thread_number_range_desc)
 230             .positional_arg(InputFileName,"input-file","input file name")
 231             .positional_arg(OutputFileName,"output-file","output file name")
 232             .positional_arg(MAX_CHAR_PER_INPUT_SLICE, "max-slice-size","the maximum number of characters in one slice")
 233             .arg(silent,"silent","no output except elapsed time")
 234             );
 235         generate_if_needed( InputFileName.c_str() );
 236
 237         if ( threads.first ) {
 238             for(int p = threads.first;  p <= threads.last; p=threads.step(p) ) {
 239                 if ( !silent ) printf("threads = %d ", p);
 240                 tbb::global_control c(tbb::global_control::max_allowed_parallelism, p);
 241                 if(!run_pipeline (p))
 242                     return 1;
 243             }
 244         } else { // Number of threads wasn't set explicitly. Run serial and parallel version
 245             { // serial run
 246                 if ( !silent ) printf("serial run   ");
 247                 tbb::global_control c(tbb::global_control::max_allowed_parallelism, 1);
 248                 if(!run_pipeline (1))
 249                     return 1;
 250             }
 251             { // parallel run (number of threads is selected automatically)
 252                 if ( !silent ) printf("parallel run ");
 253                 tbb::global_control c(tbb::global_control::max_allowed_parallelism, utility::get_default_num_threads());
 254                 if(!run_pipeline (utility::get_default_num_threads()))
 255                     return 1;
 256             }
 257         }
 258
 259         utility::report_elapsed_time((tbb::tick_count::now() - mainStartTime).seconds());
 260
 261         return 0;
 262     } catch(std::exception& e) {
 263         std::cerr<<"error occurred. error text is :\"" <<e.what()<<"\"\n";
 264         return 1;
 265     }
 266 }