sort_file.hpp

Go to the documentation of this file.

/*
 * Copyright 2008,2009,2010,2011,2012 Stephan Ossowski, Korbinian Schneeberger,
 * Felix Ott, Joerg Hagmann, Alf Scotland, Sebastian Bender
 * 
 * This file is part of SHORE.
 * 
 * SHORE is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 * 
 * SHORE is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 * 
 * You should have received a copy of the GNU General Public License
 * along with SHORE.  If not, see <http://www.gnu.org/licenses/>.
 */


#ifndef SHORE_ALGO_SORT_FILE_HPP__
#define SHORE_ALGO_SORT_FILE_HPP__

#include <unistd.h>
#include <fcntl.h>
#include <sys/mman.h>
#include <sys/types.h>
#include <sys/stat.h>


#include <algorithm>
#include <cerrno>
#include <cstring>
#include <iostream>
#include <iterator>
#include <list>
#include <map>
#include <set>
#include <vector>
#include <string>

#include <boost/filesystem.hpp>

namespace bfs=boost::filesystem;

#include "shore/base/memops.hpp"
#include "shore/base/util.hpp"
#include "shore/base/stringops.hpp"
#include "shore/base/pathops.hpp"
#include "shore/container/helpers.hpp"
#include "shore/stream/streams.hpp"
#include "shore/stream/stream_slicer.hpp"

namespace shore {

bool cmp_field0(const char*const c0,const char*const c1);

bool scmp_field0(const std::string &s0,const std::string &s1);

bool cmp_field0_lex(const char*const c0,const char*const c1);

bool cmp_line(const char*const c0,const char*const c1);

bool cmp_maplist(const char*const c0,const char*const c1);

bool cmp_maplist_readid(const char*const c0,const char*const c1);

bool cmp_sam(const char*const c0,const char*const c1);

template<int K>
bool cmp_field(const char*const c0,const char*const c1)
{
        // find the (K+1)'th field and then compare
        const char* nc0=c0;
        const char* nc1=c1;

        int ntab=0;
        while(ntab<K)
        {
                if(nc0[0]=='\n') break;
                if(nc0[0]=='\t') ++ntab;
                ++nc0;
        }
        ntab=0;
        while(ntab<K)
        {
                if(nc1[0]=='\n') break;
                if(nc1[0]=='\t') ++ntab;
                ++nc1;
        }

        return cmp_field0(nc0,nc1);
}

template<int K>
bool cmp_field_lex(const char*const c0,const char*const c1)
{
        // find the (K+1)'th field and then compare
        const char* nc0=c0;
        const char* nc1=c1;

        int ntab=0;
        while(ntab<K)
        {
                if(nc0[0]=='\n') break;
                if(nc0[0]=='\t') ++ntab;
                ++nc0;
        }
        ntab=0;
        while(ntab<K)
        {
                if(nc1[0]=='\n') break;
                if(nc1[0]=='\t') ++ntab;
                ++nc1;
        }

        return cmp_field0_lex(nc0,nc1);
}


class cmp_tokens
{
public:

        enum CRes
        {
                C_LT=0,
                C_EQ=1,
                C_GT=2
        };

        struct rowpair
        {
                const char* c0;
                const char* c1;

                rowpair(const char*const p0,const char*const p1);
        };

        typedef CRes(*Ord_t)(rowpair& rp);

        static CRes c_ign(rowpair& rp);

        static CRes c_text(rowpair& rp);

        static CRes c_uint(rowpair& rp);

        static CRes c_int(rowpair& rp);

        static CRes c_ufloat(rowpair& rp);

        static CRes c_float(rowpair& rp);

private:

        std::vector<Ord_t> m_vo;
        std::vector<bool> m_vr;
        std::vector<size_t> m_vc;
        std::set<size_t> m_sc;
        std::vector<CRes> m_res;

        bool m_commentsfirst;

public:

        cmp_tokens();

        void set_commentsfirst(const bool b);
        
        cmp_tokens& add_key(
          const size_t column,const Ord_t o,const bool reverse);

        const std::set<size_t>& get_columns() const;

        const std::vector<size_t>& get_columnsv() const;

        bool operator()(const char*const c0,const char*const c1);

        bool operator()(const std::string& s0,const std::string& s1);
};


template<class StrictWeakOrdering=bool(*)(const char*,const char*)>
class line_sorter
{
 private:

        StrictWeakOrdering m_cmp;

        std::string m_temp;

        size_t m_blocksizelimit;

        size_t m_maxopenfiles;

        // discard redundant lines when sorting / merging
        bool m_uniq;
        // fail instead of discarding if uniq is set & lines are redundant
        bool m_assertuniq;

        bool m_safemerge;
        bool m_mmapmerge;

        std::vector<const char*> m_linebuf;


        void merge_files_rec(const std::vector<std::string>& infiles,
                             std::ostream& out)
        {
                std::multimap<long,std::string> inf;
                for(size_t i=0;i<infiles.size();++i)
                {
                        long fs=shore::istreams(infiles[i]).get_streamsize(0);
                        if(fs<0) fs=std::numeric_limits<long>::max();

                        inf.insert(std::make_pair(fs,infiles[i]));
                }

                shore::ostreams tmpout;

                // merge as few and as small as possible files to intermediate tempfile
                while(inf.size()>m_maxopenfiles)
                {
                        std::vector<std::string> merfn;

                        const size_t nfiles=
                                std::min(m_maxopenfiles,inf.size()-m_maxopenfiles+1);
                        
                        while(merfn.size()<nfiles)
                        {
                                merfn.push_back(inf.begin()->second);
                                inf.erase(inf.begin());
                        }

                        tmpout.open_tempfile(m_temp+"/shore-sort_merge");
                        merge_files(merfn,tmpout.back());
                        tmpout.close_all();

                        long fs=shore::istreams(tmpout.get_streamname(tmpout.size()-1)).get_streamsize(0);
                        if(fs<0)
                                fs=std::numeric_limits<long>::max();

                        inf.insert(std::make_pair(fs,tmpout.get_streamname(tmpout.size()-1)));
                }

                // final merge
                merge_files(shore::map_values(inf),out);
        }

        void sort_slices(shore::stream_slicer &strcut,std::ostream& out,
                         const std::string &name)
        {
                bool incomplete_line=false;

                shore::ostreams tmpout;

                while(strcut.has_data())
                {
                        const char*const data_begin=strcut.current().data;
                        const char*const data_end=data_begin+strcut.current().size;

                        // complete a line larger than blocksizelimit
                        if(incomplete_line)
                        {
                                const char* first_eol=
                                        std::find(data_begin,data_end,'\n');

                                if(first_eol!=data_end)
                                {
                                        incomplete_line=false;
                                        ++first_eol;
                                }

                                tmpout.back().write(data_begin,first_eol-data_begin);
                                strcut.move(first_eol-data_begin);

                                if(!incomplete_line)
                                        tmpout.close_all();
                        }
                        // if this is the last block
                        else if(strcut.current().eof)
                        {
                                // this is the only block and uniq is not set
                                // -> direct output
                                if((strcut.current().offset==0)&&(!m_uniq))
                                        sort_block(data_begin,data_end,out);
                                // this is the last of multiple blocks or uniq is set
                                // -> write to temp file
                                else
                                {
                                        tmpout.open_tempfile(m_temp
                                                             +"/shore-sort_"+name);
                                        sort_block(data_begin,data_end,tmpout.back());
                                        tmpout.close_all();
                                }
                                strcut.move(strcut.current().size);
                        }
                        // not the last block 
                        // -> find the last eol of the block before sort
                        // -> write to temp files
                        else
                        {
                                std::reverse_iterator<const char*> rbeg(data_end);
                                std::reverse_iterator<const char*> rend(data_begin);

                                const char*const last_eol=std::find(rbeg,rend,'\n').base();

                                if(last_eol!=data_begin)
                                {
                                        tmpout.open_tempfile(m_temp
                                                             +"/shore-sort_"+name);
                                        sort_block(data_begin,last_eol,tmpout.back());
                                        tmpout.close_all();
                                        strcut.move(last_eol-data_begin);
                                }
                                // handle lines larger than blocksizelimit (no eol found)
                                else
                                {
                                        tmpout.open_tempfile(m_temp
                                                             +"/shore-sort_"+name);

                                        tmpout.back().write(data_begin,
                                                            strcut.current().size);
                                        strcut.move(strcut.current().size);

                                        incomplete_line=true;
                                }
                        }
                }

                // reset the linebuf capacity to default
                {
                        std::vector<const char*> tmp;
                        m_linebuf.swap(tmp);
                }

                // merge temp files
                if(tmpout.size()!=0)
                {
                        std::vector<std::string> fnams(tmpout.size());
                        for(size_t i=0;i<tmpout.size();++i)
                                fnams[i]=tmpout.get_streamname(i);
                        merge_files(fnams,out);
                }
        }

public:

        line_sorter(StrictWeakOrdering cmp,
                    const std::string& tmp=".",
                    const double max_blockMB=2048.0,
                    const bool uniq=false,
                    const bool assertuniq=false,
                    const size_t maxopenfiles=0)
        :m_cmp(cmp),
         m_temp(tmp.empty()?".":tmp),
         m_blocksizelimit(static_cast<size_t>(std::ceil(max_blockMB*(1<<20)))),
         m_maxopenfiles(maxopenfiles),
         m_uniq(uniq),
         m_assertuniq(assertuniq),
         m_safemerge(true),
         m_mmapmerge(false)
        {
                if(m_maxopenfiles==0)
                {
                        try
                        {
                                m_maxopenfiles=shore::get_maxopenfiles()>>1;
                        }
                        catch(const std::exception &e)
                        {
                                std::cerr<<"# WARNING: "<<e.what()<<std::endl;
                                std::cerr<<"# maxopenfiles set to 128"<<std::endl;
                                m_maxopenfiles=128;
                        }
                }
        }

        line_sorter &safe()
        {
                m_safemerge=true;
                return *this;
        }

        line_sorter &unsafe()
        {
                m_safemerge=false;
                return *this;
        }

        line_sorter &mmapmerge()
        {
                m_mmapmerge=true;
                return *this;
        }

        line_sorter &nommapmerge()
        {
                m_mmapmerge=false;
                return *this;
        }

        void sort_block(const char*const data_begin,const char*const data_end,
                        std::ostream& out)
        {
                m_linebuf.push_back(data_begin);

                const char* eol=std::find(data_begin,data_end,'\n');
                while(eol<(data_end-1))
                {
                        m_linebuf.push_back(eol+1);
                        eol=std::find(eol+1,data_end,'\n');
                }

                // ensure that the last line is also terminated by '\n'
                const bool eofnewline=((*(data_end-1))=='\n');
                const std::string lastline=eofnewline?
                        "":(std::string(m_linebuf.back(),data_end)+'\n');

                if(!eofnewline)
                        m_linebuf.back()=lastline.c_str();

                std::sort(m_linebuf.begin(),m_linebuf.end(),m_cmp);

                typedef std::vector<const char*>::const_iterator i_t;
                for(i_t i=m_linebuf.begin();i!=m_linebuf.end();++i)
                {
                        const char*const s=(*i);
                        const char* nl=s;

                        while((*nl)!='\n')
                                ++nl;

                        if(eofnewline||(s!=m_linebuf.back()))
                                ++nl;

                        out.write(s,nl-s);
                }

                m_linebuf.clear();
        }

        void sort_file(const std::string& fn_in,std::ostream& out)
        {
                // Cut the file into slices.
                shore::stream_slicer strcut(fn_in,m_blocksizelimit);

                sort_slices(strcut,out,shore::path_leaf(fn_in));

                /*
                // decompress if necessary
                if(shore::istreams::is_gzip(fn_in)||shore::istreams::is_xz(fn_in))
                {
                        shore::ostreams tmp;
                        tmp.open_tempfile(m_temp+"/shore-sort_"+shore::path_leaf(fn_in),
                                          shore::ostreams::CODEC_PLAIN);

                        {
                                shore::istreams in(fn_in);
                                if(in[0].peek()!=EOF)
                                        tmp[0]<<in[0].rdbuf();
                                in.read_error_check_rdbuf(0,"line_sorter::sort_file("+fn_in+")");
                        }

                        tmp.close_all();
                        sort_file(tmp.get_streamname(0),out);
                        return;
                }

                // open & stat input file
                shore::mmapping mmpp(fn_in);
                const size_t total_space=mmpp.get_filesize();

                bool incomplete_line=false;

                size_t processed_space=0;

                shore::ostreams tmpout;

                while(processed_space<total_space)
                {
                        const size_t block_space=
                                std::min(m_blocksizelimit,total_space-processed_space);

                        mmpp.map(processed_space,block_space);

                        const char*const data_begin=mmpp.begin();
                        const char*const data_end=mmpp.end();

                        // complete a line larger than blocksizelimit
                        if(incomplete_line)
                        {
                                const char* first_eol=
                                        std::find(data_begin,data_end,'\n');

                                if(first_eol!=data_end)
                                {
                                        incomplete_line=false;
                                        ++first_eol;
                                }

                                tmpout.back().write(data_begin,first_eol-data_begin);
                                processed_space+=(first_eol-data_begin);

                                if(!incomplete_line) tmpout.close_all();
                        }
                        // if this is the last block
                        else if((processed_space+block_space)==total_space)
                        {
                                // this is the only block and uniq is not set
                                // -> direct output
                                if((processed_space==0)&&(!m_uniq))
                                {
                                        sort_block(data_begin,data_end,out);
                                }
                                // this is the last of multiple blocks or uniq is set
                                // -> write to temp file
                                else
                                {
                                        tmpout.open_tempfile(m_temp+"/shore-sort_"+shore::path_leaf(fn_in));
                                        sort_block(data_begin,data_end,tmpout.back());
                                        tmpout.close_all();
                                }
                                processed_space+=block_space;
                        }
                        // not the last block 
                        // -> find the last eol of the block before sort
                        // -> write to temp files
                        else
                        {
                                std::reverse_iterator<const char*> rbeg(data_end);
                                std::reverse_iterator<const char*> rend(data_begin);

                                const char*const last_eol=std::find(rbeg,rend,'\n').base();

                                if(last_eol!=data_begin)
                                {
                                        tmpout.open_tempfile(m_temp+"/shore-sort_"
                                                             +shore::path_leaf(fn_in));
                                        sort_block(data_begin,last_eol,tmpout.back());
                                        tmpout.close_all();
                                        processed_space+=(last_eol-data_begin);
                                }
                                // handle lines larger than blocksizelimit (no eol found)
                                else
                                {
                                        tmpout.open_tempfile(m_temp+"/shore-sort_"
                                                             +shore::path_leaf(fn_in));

                                        tmpout.back().write(data_begin,block_space);
                                        processed_space+=block_space;

                                        incomplete_line=true;
                                }
                        }
                }

                mmpp.unmap();

                // reset the linebuf capacity to default
                {
                        std::vector<const char*> tmp;
                        m_linebuf.swap(tmp);
                }

                // merge temp files
                if(tmpout.size()!=0)
                {
                        std::vector<std::string> fnams(tmpout.size());
                        for(size_t i=0;i<tmpout.size();++i)
                                fnams[i]=tmpout.get_streamname(i);
                        merge_files(fnams,out);
                }
                */
        }

        void sort_file(std::istream &is,std::ostream& out)
        {
                // Cut the stream into slices.
                shore::stream_slicer strcut(is,m_blocksizelimit);

                sort_slices(strcut,out,"stream");
        }

        void sort_file(const std::string& fn_in,const std::string& fn_out)
        {
                shore::ostreams out;

                out.open_file(fn_out);

                sort_file(fn_in,out[0]);

                out.close_all();
        }

        void sort_file(const std::string& fn)
        {
                shore::ostreams to;

                to.open_tempfile(fn,fn,shore::istreams::get_codec(fn),
                                 shore::ostreams::EXISTS_REPLACE);

                sort_file(fn,to.back());

                // automatically replaces fn
                to.close_all();
        }

        void sort_files(const std::vector<std::string>& fn,std::ostream& out)
        {
                if(fn.empty())
                        return;

                shore::ostreams tmps;
                std::vector<std::string> fnams(fn.size());
                for(size_t i=0;i<fn.size();++i)
                {
                        tmps.open_tempfile(m_temp+"/shore-sort_"+shore::path_leaf(fn[i]));
                        sort_file(fn[i],tmps.back());
                        tmps.close_stream(i);
                        fnams[i]=tmps.get_streamname(i);
                }
                merge_files(fnams,out);
        }

        void sort_files(
          const std::vector<std::string>& fn,const std::string& fn_out)
        {
                shore::ostreams os(fn_out);
                sort_files(fn,os[0]);
                os.close_all();
        }

        bool is_sorted(std::istream& in)
        {
                std::string buf,buf_last;
                std::getline(in,buf_last);
                buf_last+='\n';

                if(!m_uniq)
                {
                        while(std::getline(in,buf))
                        {
                                buf+='\n';
                                if(m_cmp(buf.c_str(),buf_last.c_str()))
                                        return false;
                                buf_last.swap(buf);
                        }
                }
                else
                {
                        while(std::getline(in,buf))
                        {
                                buf+='\n';
                                if(!m_cmp(buf_last.c_str(),buf.c_str()))
                                        return false;
                                buf_last.swap(buf);
                        }
                }
                shore::istreams::read_error_check(in,"<stream>","line_sorter::is_sorted (1)");
                return true;
        }

        bool is_sorted(std::istream& in,size_t*const line)
        {
                std::string buf,buf_last;
                std::getline(in,buf_last);
                buf_last+='\n';

                ++(*line);

                if(!m_uniq)
                {
                        while(std::getline(in,buf))
                        {
                                buf+='\n';
                                if(m_cmp(buf.c_str(),buf_last.c_str()))
                                        return false;
                                buf_last.swap(buf);
                                ++(*line);
                        }
                }
                else
                {
                        while(std::getline(in,buf))
                        {
                                buf+='\n';
                                if(!m_cmp(buf_last.c_str(),buf.c_str()))
                                        return false;
                                buf_last.swap(buf);
                                ++(*line);
                        }
                }
                shore::istreams::read_error_check(in,"<stream>","line_sorter::is_sorted (2)");
                return true;
        }

        bool is_sorted(const std::string& fn,size_t*const line=0)
        {
                shore::istreams in(fn);

                if(line==0)
                        return is_sorted(in.back());
                return is_sorted(in.back(),line);
        }

        void merge_files(
          const std::vector<std::string>& infiles,std::ostream& out)
        {
                if(infiles.empty())
                        return;

                // do a step-wise merge if the number of files is too high
                if((m_maxopenfiles>1)&&(infiles.size()>m_maxopenfiles))
                {
                        merge_files_rec(infiles,out);
                }
                // actual file merger part
                else
                {
                        shore::istreams in;
                        // mmap files to save file descriptors?
                        in.use_mmap(m_mmapmerge);
                        in.open_files(infiles.begin(),infiles.end());
                        
                        std::multimap<const char*,size_t,StrictWeakOrdering> filemap(m_cmp);

                        std::vector<std::string> buffers(in.size());
                        std::string auxbuf;

                        // build map
                        for(size_t i=0;i<in.size();++i)
                        {
                                if(std::getline(in[i],buffers[i]))
                                {
                                        buffers[i]+='\n';
                                        const char*const ptr=buffers[i].c_str();
                                        filemap.insert(std::make_pair(ptr,i));
                                }
                                else
                                        in.read_error_check(i,"line_sorter::merge_files (1)");
                        }

                        bool assertinit=true;

                        if(m_uniq&&(!filemap.empty()))
                        {
                                auxbuf=filemap.begin()->first;
                                out<<filemap.begin()->first;
                        }

                        // merge
                        while(!filemap.empty())
                        {
                                if(!m_uniq||m_cmp(auxbuf.c_str(),filemap.begin()->first))
                                {
                                        out<<filemap.begin()->first;
                                }
                                else if(m_assertuniq)
                                {
                                        if(assertinit)
                                                assertinit=false;
                                        else
                                                throw std::runtime_error(
                                                        "merge_files: line is not unique {\n"
                                                        +auxbuf+filemap.begin()->first+"}");
                                }

                                const size_t fileid=filemap.begin()->second;

                                filemap.erase(filemap.begin());

                                if(m_safemerge||m_uniq)
                                        buffers[fileid].swap(auxbuf);

                                if(std::getline(in[fileid],buffers[fileid]))
                                {
                                        buffers[fileid]+='\n';
                                        const char*const ptr=buffers[fileid].c_str();

                                        // check sorting
                                        if(m_safemerge&&m_cmp(ptr,auxbuf.c_str()))
                                                throw std::runtime_error(
                                                        "merge_files: "+in.get_streamname(fileid)
                                                        +" is not sorted appropriately {"
                                                        +auxbuf.substr(0,80)+" ...}");

                                        filemap.insert(std::make_pair(ptr,fileid));
                                }
                                else if(!in[fileid].eof())
                                        in.read_error_check(fileid,"line_sorter::merge_files (2)");
                        }
                }
        }

        void merge_files(
          const std::vector<std::string>& infiles,const std::string& fn_out)
        {
                shore::ostreams out(fn_out);
                merge_files(infiles,out[0]);
                out.close_all();
        }

        std::pair<off_t,std::string> upper_bound(std::istream& in,
                                                 const std::string& what,
                                                 const off_t rangemin=0l,
                                                 off_t rangemax=-1l)
        {
                const std::string nlwhat=what+'\n';
                std::string buf;

                if(rangemax==-1l)
                {
                        in.seekg(0,std::ios::end);

                        if(!in.good())
                                throw std::runtime_error("shore::upper_bound: seek failed");

                        rangemax=in.tellg();

                        if(rangemax<0l)
                                throw std::runtime_error("shore::upper_bound: tell failed");
                }

                const off_t spos=rangemin+((rangemax-rangemin)>>1);

                in.seekg(spos);
                if(!in.good())
                        throw std::runtime_error("shore::upper_bound: seek failed");

                std::getline(in,buf);
                buf+='\n';

                if(!in)
                {
                        if(in.eof())
                                in.clear();
                        else
                                shore::istreams::read_error_check(in,"<stream>",
                                                                  "line_sorter::upper_bound (1)");
                }

                off_t p0=in.tellg();
                if(p0<0l)
                        throw std::runtime_error("shore::upper_bound: tell failed");

                if(p0>=rangemax)
                {
                        p0=rangemin;
                        in.seekg(p0);
                }
                std::getline(in,buf);
                buf+='\n';

                if(!in)
                {
                        if(in.eof())
                                in.clear();
                        else
                                shore::istreams::read_error_check(in,"<stream>",
                                                                  "line_sorter::upper_bound (2)");
                }

                off_t p1=in.tellg();
                if(p1<0l)
                        throw std::runtime_error("shore::upper_bound: tell failed");

                if((p1>=rangemax)&&(p0!=rangemin))
                {
                        p0=rangemin;
                        in.seekg(p0);

                        std::getline(in,buf);
                        buf+='\n';

                        if(!in)
                        {
                                if(in.eof())
                                        in.clear();
                                else
                                        shore::istreams::read_error_check(in,"<stream>",
                                                                          "line_sorter::upper_bound (3)");
                        }

                        p1=in.tellg();
                        if(p1<0l)
                                throw std::runtime_error("shore::upper_bound: tell failed");
                }

                if(p1>=rangemax)
                {
                        if(m_cmp(nlwhat.c_str(),buf.c_str()))
                        {
                                in.seekg(p0);
                                return std::make_pair(p0,buf.substr(0,buf.size()-1));
                        }
                        else
                                return std::make_pair(p1,"");
                }

                // if buf compares greater than what
                if(m_cmp(nlwhat.c_str(),buf.c_str()))
                {
                        return upper_bound(in,what,rangemin,p1);
                }
                else
                {
                        return upper_bound(in,what,p1,rangemax);
                }
        }

        std::pair<off_t,std::string> lower_bound(std::istream& in,
          const std::string& what,const off_t rangemin=0l,off_t rangemax=-1l)
        {
                const std::string nlwhat=what+'\n';
                std::string buf;

                if(rangemax==-1l)
                {
                        in.seekg(0,std::ios::end);

                        if(!in.good())
                                throw std::runtime_error("shore::lower_bound: seek failed");

                        rangemax=in.tellg();

                        if(rangemax<0l)
                                throw std::runtime_error("shore::lower_bound: tell failed");
                }

                const off_t spos=rangemin+((rangemax-rangemin)>>1);

                in.seekg(spos);
                if(!in.good())
                        throw std::runtime_error("shore::lower_bound: seek failed");

                std::getline(in,buf);
                buf+='\n';

                if(!in)
                {
                        if(in.eof())
                                in.clear();
                        else
                                shore::istreams::read_error_check(in,"<stream>",
                                                                  "line_sorter::lower_bound (1)");
                }

                off_t p0=in.tellg();
                if(p0<0l)
                        throw std::runtime_error("shore::lower_bound: tell failed");

                if(p0>=rangemax)
                {
                        p0=rangemin;
                        in.seekg(p0);
                }
                std::getline(in,buf);
                buf+='\n';

                if(!in)
                {
                        if(in.eof())
                                in.clear();
                        else
                                shore::istreams::read_error_check(in,"<stream>",
                                                                  "line_sorter::lower_bound (2)");
                }

                off_t p1=in.tellg();
                if(p1<0l)
                        throw std::runtime_error("shore::lower_bound: tell failed");

                if((p1>=rangemax)&&(p0!=rangemin))
                {
                        p0=rangemin;
                        in.seekg(p0);

                        std::getline(in,buf);
                        buf+='\n';

                        if(!in)
                        {
                                if(in.eof())
                                        in.clear();
                                else
                                        shore::istreams::read_error_check(in,"<stream>",
                                                                          "line_sorter::lower_bound (3)");
                        }

                        p1=in.tellg();
                        if(p1<0l)
                                throw std::runtime_error("shore::lower_bound: tell failed");
                }

                if(p1>=rangemax)
                {
                        if(!m_cmp(buf.c_str(),nlwhat.c_str()))
                        {
                                in.seekg(p0);
                                return std::make_pair(p0,buf.substr(0,buf.size()-1));
                        }
                        else
                                return std::make_pair(p1,"");
                }

                // if buf does not compare less than what
                if(!m_cmp(buf.c_str(),nlwhat.c_str()))
                {
                        return lower_bound(in,what,rangemin,p1);
                }
                else
                {
                        return lower_bound(in,what,p1,rangemax);
                }
        }
};

} // namespace

#endif // SHORE_ALGO_SORT_FILE_HPP__