10.4 在多线程环境中使用传统 iostream

stream_MT

多线程安全类的基类。 

streambuf

缓冲区的基类。 

unsafe_ios

该类包含各种流类通用的状态变量；例如，错误和格式化状态。 

unsafe_istream

该类支持从 streambuf 检索的字符序列的有格式和无格式转换。

unsafe_ostream

该类支持存储到 streambuf 中的字符序列的有格式和无格式转换。

unsafe_iostream

该类合并 unsafe_istream 类和 unsafe_ostream 类，以便进行双向操作。

class streambuf: public stream_MT {...};
class ios: virtual public unsafe_ios, public stream_MT {...};
class istream: virtual public ios, public unsafe_istream {...};

char *oct_r (char *buf,

int buflen,

long num,

int width)

将指针返回到用八进制表示数字的 ASCII 字符串。非零宽度假定为格式化的字段宽度。返回值不保证指向用户提供缓冲区的开始部分。 

char *hex_r (char *buf,

int buflen,

long num,

int width)

将指针返回到用十六进制表示数字的 ASCII 字符串。非零宽度假定为格式化的字段宽度。返回值不保证指向用户提供缓冲区的开始部分。 

char *dec_r (char *buf,

int buflen,

long num,

int width)

将指针返回到用十进制表示数字的 ASCII 字符串。非零宽度假定为格式化的字段宽度。返回值不保证指向用户提供缓冲区的开始部分。 

char *chr_r (char *buf,

int buflen,

long num,

int width)

返回指向包含字符 chr 的 ASCII 字符串的指针。如果宽度非零，则字符串包含后跟 chr 的 width 个空格。返回值不保证指向用户提供缓冲区的开始部分。

char *form_r (char *buf,

int buflen,

long num,

int width)

返回由 sprintf 格式化字符串的指针，其中使用了格式字符串 format 和其余参数。缓冲区必须具有足够的空间以包含格式化的字符串。

#include <iostream.h>
enum iostate {IOok, IOeof, IOfail};

iostate read_number(istream& istr, int& num)
{
    stream_locker sl(istr, stream_locker::lock_now);

    istr >> num;

    if (istr.eof()) return IOeof;
    if (istr.fail()) return IOfail;
    return IOok;
}

#include <iostream.h>
#include <rlocks.h>
void fetch_line(istream& istr, char* line, int& linecount)
{
    stream_locker sl(istr, stream_locker::lock_defer);

    sl.lock(); // lock the stream istr
    istr >> line;
    linecount = istr.gcount();
    sl.unlock(); // unlock istr
    ...
}

#include <rlocks.h>
#include <iostream.h>
class mystream: public istream {

    // other definitions...
    int getRecord(char* name, int& id, float& gpa);
};

int mystream::getRecord(char* name, int& id, float& gpa)
{
    stream_locker sl(this, stream_locker::lock_now);

    *this >> name;
    *this >> id;
    *this >> gpa;

    return this->fail() == 0;
}

    cout.unsafe_ostream::put(’4’);

    cin.unsafe_istream::read(buf, len);

#include <iostream.h>
//disable mt-safety
cout.set_safe_flag(stream_MT::unsafe_object);    
//disable mt-safety
cin.set_safe_flag(stream_MT::unsafe_object);    
cout.put(”4’);
cin.read(buf, len);

fs.set_safe_flag(stream_MT::unsafe_object);// disable MT-safety
    .... do various i/o operations

    generic_lock();
    fs.set_safe_flag(stream_MT::unsafe_object);
    ... do various i/o operations
    generic_unlock();

    stream_MT
    stream_locker
    unsafe_ios
    unsafe_istream
    unsafe_ostream
    unsafe_iostream
    unsafe_fstreambase
    unsafe_strstreambase

class streambuf: public stream_MT {...};
class unsafe_ios {...};
class ios: virtual public unsafe_ios, public stream_MT {...};
class unsafe_fstreambase: virtual public unsafe_ios {...};
class fstreambase: virtual public ios, public unsafe_fstreambase
  {...};
class unsafe_strstreambase: virtual public unsafe_ios {...};
class strstreambase: virtual public ios, public unsafe_strstreambase {...};
class unsafe_istream: virtual public unsafe_ios {...};
class unsafe_ostream: virtual public unsafe_ios {...};
class istream: virtual public ios, public unsafe_istream {...};
class ostream: virtual public ios, public unsafe_ostream {...};
class unsafe_iostream: public unsafe_istream, public unsafe_ostream {...};

 class streambuf {
 public:
   int sgetc_unlocked();                
   void sgetn_unlocked(char *, int);
   int snextc_unlocked();
   int sbumpc_unlocked();
   void stossc_unlocked();
   int in_avail_unlocked();
   int sputbackc_unlocked(char);
   int sputc_unlocked(int);
   int sputn_unlocked(const char *, int);
   int out_waiting_unlocked();
 protected:
   char* base_unlocked();
   char* ebuf_unlocked();
   int blen_unlocked();
   char* pbase_unlocked();
   char* eback_unlocked();
   char* gptr_unlocked();
   char* egptr_unlocked();
   char* pptr_unlocked();
   void setp_unlocked(char*, char*);
   void setg_unlocked(char*, char*, char*);
   void pbump_unlocked(int);
   void gbump_unlocked(int);
   void setb_unlocked(char*, char*, int);
   int unbuffered_unlocked();
   char *epptr_unlocked();
   void unbuffered_unlocked(int);
   int allocate_unlocked(int);
 };

 class filebuf: public streambuf {
 public:
  int is_open_unlocked();
  filebuf* close_unlocked();
  filebuf* open_unlocked(const char*, int, int =
    filebuf::openprot);

  filebuf* attach_unlocked(int);
 };

 class strstreambuf: public streambuf {
 public:
  int freeze_unlocked();
  char* str_unlocked();
 };


 unsafe_ostream& endl(unsafe_ostream&);
 unsafe_ostream& ends(unsafe_ostream&);
 unsafe_ostream& flush(unsafe_ostream&);
 unsafe_istream& ws(unsafe_istream&);
 unsafe_ios& dec(unsafe_ios&);
 unsafe_ios& hex(unsafe_ios&);
 unsafe_ios& oct(unsafe_ios&);

 char* dec_r (char* buf, int buflen, long num, int width)
 char* hex_r (char* buf, int buflen, long num, int width)
 char* oct_r (char* buf, int buflen, long num, int width)
 char* chr_r (char* buf, int buflen, long chr, int width)
 char* str_r (char* buf, int buflen, const char* format, int width
    = 0);
 char* form_r (char* buf, int buflen, const char* format,...)

    int c;
    cin.get(c);

cout << " Error message:" << errstring[err_number] << "\n";

#include <fstream.h>
#include <rlocks.h>

void lock_example (fstream& fs)
{
    const int len = 128;
    char buf[len];
    int offset = 48;
    stream_locker s_lock(fs, stream_locker::lock_now);
    .....// open file
    fs.seekg(offset, ios::beg);
    fs.read(buf, len);
}

{
    ...
    stream_locker file_lck(openfile_stream,
                             stream_locker::lock_defer);
    ....
    file_lck.lock();  // lock openfile_stream
    openfile_stream << "Value: " << int_value << "\n";
    if(!openfile_stream) {
            file_error("Output of value failed\n");
            return;
    }
    file_lck.unlock(); // unlock openfile_stream
}

#include <fstream.h>
#include <thread.h>
fstream* fp;

void *process_rtn(void*)
{
    // body of sub-threads which uses fp...
}

void multi_process(const char* filename, int numthreads)
{
    fp = new fstream(filename, ios::in); // create fstream object
                                         // before creating threads.
    // create threads
    for (int i=0; i<numthreads; i++)
            thr_create(0, STACKSIZE, process_rtn, 0, 0, 0);

        ...
    // wait for threads to finish
    for (int i=0; i<numthreads; i++)
            thr_join(0, 0, 0);

    delete fp;                          // delete fstream object after
    fp = NULL;                         // all threads have completed.
}

// create tagged thread data
// the output file is of the form:
//         <tag><string of data>\n
// where tag is an integer value in a unsigned char.
// Allows up to 255 threads to be run in this application
// <string of data> is any printable characters
// Because tag is an integer value written as char,
// you need to use od to look at the output file, suggest:
//            od -c out.file |more

#include <stdlib.h>
#include <stdio.h>
#include <iostream.h>
#include <fstream.h>
#include <thread.h>

struct thread_args {
  char* filename;
  int thread_tag;
};

const int thread_bufsize = 256;

// entry routine for each thread
void* ThreadDuties(void* v) {
// obtain arguments for this thread
  thread_args* tt = (thread_args*)v;
  char ibuf[thread_bufsize];
  // open thread input file
  ifstream instr(tt->filename);
  stream_locker lockout(cout, stream_locker::lock_defer);
  while(1) {
  // read a line at a time
    instr.getline(ibuf, thread_bufsize - 1, ’\n’);
    if(instr.eof())
      break;
  // lock cout stream so the i/o operation is atomic
    lockout.lock();
  // tag line and send to cout
    cout << (unsigned char)tt->thread_tag << ibuf << "\n";
    lockout.unlock();
  }
  return 0;
}

int main(int argc, char** argv) {
  // argv: 1+ list of filenames per thread
   if(argc < 2) {
     cout << “usage: " << argv[0] << " <files..>\n";
     exit(1);
   }
  int num_threads = argc - 1;
  int total_tags = 0;

// array of thread_ids
  thread_t created_threads[thread_bufsize];
// array of arguments to thread entry routine
  thread_args thr_args[thread_bufsize];
  int i;
  for(i = 0; i < num_threads; i++) {
    thr_args[i].filename = argv[1 + i];
// assign a tag to a thread - a value less than 256
    thr_args[i].thread_tag = total_tags++;
// create threads
    thr_create(0, 0, ThreadDuties, &thr_args[i],
           THR_SUSPENDED, &created_threads[i]);
  }

  for(i = 0; i < num_threads; i++) {
    thr_continue(created_threads[i]);
  }
  for(i = 0; i < num_threads; i++) {
    thr_join(created_threads[i], 0, 0);
  }
  return 0;
}