8372921153

Committed 21 Mar 2024 09:51AM UTC coverage: 59.234% (-0.002%) from 59.236%

Build # 8372921153

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #13965 from rgacogne/auth-ssql-unique-ptr

auth: Wrap SSql pointers in a unique pointer earlier

Run Details

32495 of 87978 branches covered (36.94%)

Branch coverage included in aggregate %.

23 of 23 new or added lines in 5 files covered. (100.0%)

22 existing lines in 8 files now uncovered.

113633 of 158717 relevant lines covered (71.59%)

3233364.78 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

86.56

/pdns/signingpipe.cc

#ifdef HAVE_CONFIG_H
#include "config.h"
#endif
#include "signingpipe.hh"
#include "misc.hh"
#include "dns_random.hh"
#include <poll.h>

#include <sys/socket.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <sched.h>

// deal with partial reads
namespace {
int readn(int fd, void* buffer, unsigned int len)
{
  unsigned int pos=0;
  int res;
  for(;;) {
    res = read(fd, (char*)buffer + pos, len - pos);
    if(res == 0) {
      if(pos)
        throw runtime_error("Signing Pipe remote shut down in the middle of a message");
      else {
        //cerr<<"Got decent EOF on "<<fd<<endl;
        return 0;
      }
    }
      
    if(res < 0) {
      if(errno == EAGAIN || errno == EINTR) {
        if(pos==0)
          return -1;
        // error handled later
        (void)waitForData(fd, -1);
        continue;
      }
      unixDie("Reading from socket in Signing Pipe loop");
    }
  
    pos+=res;
    if(pos == len)
      break;
  }
  return len;
}
}

void* ChunkedSigningPipe::helperWorker(ChunkedSigningPipe* csp, int fd)
try {
  csp->worker(fd);
  return nullptr;
}
catch(...) {
  g_log<<Logger::Error<<"Unknown exception in signing thread occurred"<<endl;
  return nullptr;
}

ChunkedSigningPipe::ChunkedSigningPipe(DNSName  signerName, bool mustSign, unsigned int workers, unsigned int maxChunkRecords)
  : d_signed(0), d_queued(0), d_outstanding(0), d_numworkers(workers), d_submitted(0), d_signer(std::move(signerName)),
    d_maxchunkrecords(maxChunkRecords), d_threads(d_numworkers), d_mustSign(mustSign), d_final(false)
{
  d_rrsetToSign = make_unique<rrset_t>();
  d_chunks.push_back(vector<DNSZoneRecord>()); // load an empty chunk
  
  if(!d_mustSign)
    return;
  
  int fds[2];
  
  for(unsigned int n=0; n < d_numworkers; ++n) {
    if(socketpair(AF_UNIX, SOCK_STREAM, 0, fds) < 0) 
      throw runtime_error("Unable to create communication socket in for ChunkedSigningPipe");
    setCloseOnExec(fds[0]);
    setCloseOnExec(fds[1]);
    d_threads[n] = std::thread(helperWorker, this, fds[1]);
    setNonBlocking(fds[0]);
    d_sockets.push_back(fds[0]);
    d_outstandings[fds[0]] = 0;
  }
}

ChunkedSigningPipe::~ChunkedSigningPipe()
{
  if(!d_mustSign)
    return;

  for(int fd :  d_sockets) {
    close(fd); // this will trigger all threads to exit
  }

  for(auto& thread : d_threads) {
    thread.join();
  }
  //cout<<"Did: "<<d_signed<<", records (!= chunks) submitted: "<<d_submitted<<endl;
}

namespace {
bool
dedupLessThan(const DNSZoneRecord& a, const DNSZoneRecord &b)
{
  return std::tuple(a.dr.getContent()->getZoneRepresentation(), a.dr.d_ttl) < std::tuple(b.dr.getContent()->getZoneRepresentation(), b.dr.d_ttl);  // XXX SLOW SLOW SLOW
}

bool dedupEqual(const DNSZoneRecord& a, const DNSZoneRecord &b)
{
  return std::tuple(a.dr.getContent()->getZoneRepresentation(), a.dr.d_ttl) == std::tuple(b.dr.getContent()->getZoneRepresentation(), b.dr.d_ttl);  // XXX SLOW SLOW SLOW
}
}

void ChunkedSigningPipe::dedupRRSet()
{
  // our set contains contains records for one type and one name, but might not be sorted otherwise
  sort(d_rrsetToSign->begin(), d_rrsetToSign->end(), dedupLessThan);
  d_rrsetToSign->erase(unique(d_rrsetToSign->begin(), d_rrsetToSign->end(), dedupEqual), d_rrsetToSign->end());
}

bool ChunkedSigningPipe::submit(const DNSZoneRecord& rr)
{
  ++d_submitted;
  // check if we have a full RRSET to sign
  if(!d_rrsetToSign->empty() && (d_rrsetToSign->begin()->dr.d_type != rr.dr.d_type ||  d_rrsetToSign->begin()->dr.d_name != rr.dr.d_name)) 
  {
    dedupRRSet();
    sendRRSetToWorker();
  }
  d_rrsetToSign->push_back(rr);
  return !d_chunks.empty() && d_chunks.front().size() >= d_maxchunkrecords; // "you can send more"
}

pair<vector<int>, vector<int> > ChunkedSigningPipe::waitForRW(bool rd, bool wr, int seconds)
{
  vector<pollfd> pfds;

  for(int & socket : d_sockets) {    
    if(d_eof.count(socket))  
      continue;
    struct pollfd pfd;
    memset(&pfd, 0, sizeof(pfd));
    pfd.fd = socket;
    if(rd)
      pfd.events |= POLLIN;
    if(wr)
      pfd.events |= POLLOUT;
    pfds.push_back(pfd);
  }

  int res = poll(&pfds[0], pfds.size(), (seconds < 0) ? -1 : (seconds * 1000)); // -1 = infinite
  if(res < 0)
    unixDie("polling for activity from signers, "+std::to_string(d_sockets.size()));
  pair<vector<int>, vector<int> > vects;
  for(auto & pfd : pfds) 
    if(pfd.revents & POLLIN)
      vects.first.push_back(pfd.fd);
    else if(pfd.revents & POLLOUT)
      vects.second.push_back(pfd.fd);
  
  return vects;
}

void ChunkedSigningPipe::addSignedToChunks(std::unique_ptr<chunk_t>& signedChunk)
{
  chunk_t::const_iterator from = signedChunk->begin();
  
  while(from != signedChunk->end()) {
    chunk_t& fillChunk = d_chunks.back();
    chunk_t::size_type room = d_maxchunkrecords - fillChunk.size();
    
    unsigned int fit = std::min(room, (chunk_t::size_type)(signedChunk->end() - from));
  
    d_chunks.back().insert(fillChunk.end(), from , from + fit);
    from+=fit;

    if(from != signedChunk->end()) // it didn't fit, so add a new chunk
      d_chunks.push_back(chunk_t());
  }
}

void ChunkedSigningPipe::sendRRSetToWorker() // it sounds so socialist!
{
  if(!d_mustSign) {
    addSignedToChunks(d_rrsetToSign);
    d_rrsetToSign->clear();
    return;
  }
  
  if(d_final && !d_outstanding) // nothing to do!
    return;
  
  bool wantRead, wantWrite;
  
  wantWrite = !d_rrsetToSign->empty();
  wantRead = d_outstanding || wantWrite;  // if we wrote, we want to read
  
  pair<vector<int>, vector<int> > rwVect;
  
  rwVect = waitForRW(wantRead, wantWrite, -1); // wait for something to happen
  
  if(wantWrite && !rwVect.second.empty()) {
    shuffle(rwVect.second.begin(), rwVect.second.end(), pdns::dns_random_engine()); // pick random available worker
    auto ptr = d_rrsetToSign.get();
    writen2(*rwVect.second.begin(), &ptr, sizeof(ptr));
    // coverity[leaked_storage]
    static_cast<void>(d_rrsetToSign.release());
    d_rrsetToSign = make_unique<rrset_t>();
    d_outstandings[*rwVect.second.begin()]++;
    d_outstanding++;
    d_queued++;
    wantWrite=false;
  } 
  
  if(wantRead) {
    while(d_outstanding) {
      for(int fd :  rwVect.first) {
        if(d_eof.count(fd))
          continue;
        
        while(d_outstanding) {
          chunk_t* chunk = nullptr;
          int res = readn(fd, &chunk, sizeof(chunk));
          if(!res) {
            if (d_outstandings[fd] > 0) {
              throw std::runtime_error("A signing pipe worker died while we were waiting for its result");
            }
            d_eof.insert(fd);
            break;
          }
          if(res < 0) {
            if(errno != EAGAIN && errno != EINTR)
              unixDie("Error reading signed chunk from thread");
            else
              break;
          }

          std::unique_ptr<rrset_t> chunkPtr(chunk);
          chunk = nullptr;
          --d_outstanding;
          d_outstandings[fd]--;
          
          addSignedToChunks(chunkPtr);
        }
      }
      if(!d_outstanding || !d_final)
        break;
      rwVect = waitForRW(true, false, -1); // wait for something to happen
    }
  }
  
  if(wantWrite) {  // our optimization above failed, we now wait synchronously
    rwVect = waitForRW(false, wantWrite, -1); // wait for something to happen
    shuffle(rwVect.second.begin(), rwVect.second.end(), pdns::dns_random_engine()); // pick random available worker
    auto ptr = d_rrsetToSign.get();
    writen2(*rwVect.second.begin(), &ptr, sizeof(ptr));
    // coverity[leaked_storage]
    static_cast<void>(d_rrsetToSign.release());
    d_rrsetToSign = make_unique<rrset_t>();
    d_outstandings[*rwVect.second.begin()]++;
    d_outstanding++;
    d_queued++;
  }
  
}

unsigned int ChunkedSigningPipe::getReady() const
{
   unsigned int sum=0; 
   for(const auto& v :  d_chunks) {
     sum += v.size(); 
   }
   return sum;
}

void ChunkedSigningPipe::worker(int fd)
try
{
  UeberBackend db("key-only");
  DNSSECKeeper dk(&db);
  
  chunk_t* chunk = nullptr;
  int res;
  for(;;) {
    res = readn(fd, &chunk, sizeof(chunk));
    if(!res)
      break;
    if(res < 0)
      unixDie("reading object pointer to sign from pdns");
    try {
      set<DNSName> authSet;
      authSet.insert(d_signer);
      addRRSigs(dk, db, authSet, *chunk);
      ++d_signed;

      writen2(fd, &chunk, sizeof(chunk));
      chunk = nullptr;
    }
    catch(const PDNSException& pe) {
      delete chunk;
      throw;
    }
    catch(const std::exception& e) {
      delete chunk;
      throw;
    }
  }
  close(fd);
}
catch(const PDNSException& pe)
{
  g_log<<Logger::Error<<"Signing thread died because of PDNSException: "<<pe.reason<<endl;
  close(fd);
}
catch(const std::exception& e)
{
  g_log<<Logger::Error<<"Signing thread died because of std::exception: "<<e.what()<<endl;
  close(fd);
}

void ChunkedSigningPipe::flushToSign()
{
  sendRRSetToWorker();
  d_rrsetToSign->clear();
}

vector<DNSZoneRecord> ChunkedSigningPipe::getChunk(bool final)
{
  if(final && !d_final) {
    // this means we should keep on reading until d_outstanding == 0
    d_final = true;
    flushToSign();
    
    for(int fd :  d_sockets) {
      shutdown(fd, SHUT_WR); // perhaps this transmits EOF the other side
      //cerr<<"shutdown of "<<fd<<endl;
    }
  }
  if(d_final)
    flushToSign(); // should help us wait
  vector<DNSZoneRecord> front=d_chunks.front();
  d_chunks.pop_front();
  if(d_chunks.empty())
    d_chunks.push_back(vector<DNSZoneRecord>());
/*  if(d_final && front.empty())
      cerr<<"getChunk returning empty in final"<<endl; */
  return front;
}



1	#ifdef HAVE_CONFIG_H
2	#include "config.h"
3	#endif
4	#include "signingpipe.hh"
5	#include "misc.hh"
6	#include "dns_random.hh"
7	#include <poll.h>
8
9	#include <sys/socket.h>
10	#include <netinet/in.h>
11	#include <netinet/tcp.h>
12	#include <sched.h>
13
14	// deal with partial reads
15	namespace {
16	int readn(int fd, void* buffer, unsigned int len)
17	{	1,667,832✔
18	unsigned int pos=0;	1,667,832✔
19	int res;	1,667,832✔
20	for(;;) {	1,667,832✔
21	res = read(fd, (char*)buffer + pos, len - pos);	1,667,506✔
22	if(res == 0) {	1,667,506✔
23	if(pos)	2,361!
24	throw runtime_error("Signing Pipe remote shut down in the middle of a message");	×
25	else {	2,361✔
26	//cerr<<"Got decent EOF on "<<fd<<endl;
27	return 0;	2,361✔
28	}	2,361✔
29	}	2,361✔
30
31	if(res < 0) {	1,665,145✔
32	if(errno == EAGAIN \|\| errno == EINTR) {	264,684!
33	if(pos==0)	264,684!
34	return -1;	264,684✔
35	// error handled later
36	(void)waitForData(fd, -1);	×
37	continue;	×
38	}	264,684✔
UNCOV 39	unixDie("Reading from socket in Signing Pipe loop");	×
UNCOV 40	}	×
41
42	pos+=res;	1,400,461✔
43	if(pos == len)	1,400,461✔
44	break;	1,400,498✔
45	}	1,400,461✔
46	return len;	1,400,787✔
47	}	1,667,832✔
48	}
49
50	void* ChunkedSigningPipe::helperWorker(ChunkedSigningPipe* csp, int fd)
51	try {	2,358✔
52	csp->worker(fd);	2,358✔
53	return nullptr;	2,358✔
54	}	2,358✔
55	catch(...) {	2,358✔
56	g_log<<Logger::Error<<"Unknown exception in signing thread occurred"<<endl;	×
57	return nullptr;	×
58	}	×
59
60	ChunkedSigningPipe::ChunkedSigningPipe(DNSName signerName, bool mustSign, unsigned int workers, unsigned int maxChunkRecords)
61	: d_signed(0), d_queued(0), d_outstanding(0), d_numworkers(workers), d_submitted(0), d_signer(std::move(signerName)),
62	d_maxchunkrecords(maxChunkRecords), d_threads(d_numworkers), d_mustSign(mustSign), d_final(false)
63	{	1,310✔
64	d_rrsetToSign = make_unique<rrset_t>();	1,310✔
65	d_chunks.push_back(vector<DNSZoneRecord>()); // load an empty chunk	1,310✔
66
67	if(!d_mustSign)	1,310✔
68	return;	524✔
69
70	int fds[2];	786✔
71
72	for(unsigned int n=0; n < d_numworkers; ++n) {	3,144✔
73	if(socketpair(AF_UNIX, SOCK_STREAM, 0, fds) < 0)	2,358!
74	throw runtime_error("Unable to create communication socket in for ChunkedSigningPipe");	×
75	setCloseOnExec(fds[0]);	2,358✔
76	setCloseOnExec(fds[1]);	2,358✔
77	d_threads[n] = std::thread(helperWorker, this, fds[1]);	2,358✔
78	setNonBlocking(fds[0]);	2,358✔
79	d_sockets.push_back(fds[0]);	2,358✔
80	d_outstandings[fds[0]] = 0;	2,358✔
81	}	2,358✔
82	}	786✔
83
84	ChunkedSigningPipe::~ChunkedSigningPipe()
85	{	1,310✔
86	if(!d_mustSign)	1,310✔
87	return;	524✔
88
89	for(int fd : d_sockets) {	2,358✔
90	close(fd); // this will trigger all threads to exit	2,358✔
91	}	2,358✔
92
93	for(auto& thread : d_threads) {	2,358✔
94	thread.join();	2,358✔
95	}	2,358✔
96	//cout<<"Did: "<<d_signed<<", records (!= chunks) submitted: "<<d_submitted<<endl;
97	}	786✔
98
99	namespace {
100	bool
101	dedupLessThan(const DNSZoneRecord& a, const DNSZoneRecord &b)
102	{	20,275✔
103	return std::tuple(a.dr.getContent()->getZoneRepresentation(), a.dr.d_ttl) < std::tuple(b.dr.getContent()->getZoneRepresentation(), b.dr.d_ttl); // XXX SLOW SLOW SLOW	20,275✔
104	}	20,275✔
105
106	bool dedupEqual(const DNSZoneRecord& a, const DNSZoneRecord &b)
107	{	6,929✔
108	return std::tuple(a.dr.getContent()->getZoneRepresentation(), a.dr.d_ttl) == std::tuple(b.dr.getContent()->getZoneRepresentation(), b.dr.d_ttl); // XXX SLOW SLOW SLOW	6,929✔
109	}	6,929✔
110	}
111
112	void ChunkedSigningPipe::dedupRRSet()
113	{	836,591✔
114	// our set contains contains records for one type and one name, but might not be sorted otherwise
115	sort(d_rrsetToSign->begin(), d_rrsetToSign->end(), dedupLessThan);	836,591✔
116	d_rrsetToSign->erase(unique(d_rrsetToSign->begin(), d_rrsetToSign->end(), dedupEqual), d_rrsetToSign->end());	836,591✔
117	}	836,591✔
118
119	bool ChunkedSigningPipe::submit(const DNSZoneRecord& rr)
120	{	844,885✔
121	++d_submitted;	844,885✔
122	// check if we have a full RRSET to sign
123	if(!d_rrsetToSign->empty() && (d_rrsetToSign->begin()->dr.d_type != rr.dr.d_type \|\| d_rrsetToSign->begin()->dr.d_name != rr.dr.d_name))	844,885✔
124	{	836,592✔
125	dedupRRSet();	836,592✔
126	sendRRSetToWorker();	836,592✔
127	}	836,592✔
128	d_rrsetToSign->push_back(rr);	844,885✔
129	return !d_chunks.empty() && d_chunks.front().size() >= d_maxchunkrecords; // "you can send more"	844,885✔
130	}	844,885✔
131
132	pair<vector<int>, vector<int> > ChunkedSigningPipe::waitForRW(bool rd, bool wr, int seconds)
133	{	727,267✔
134	vector<pollfd> pfds;	727,267✔
135
136	for(int & socket : d_sockets) {	2,181,798✔
137	if(d_eof.count(socket))	2,181,798✔
138	continue;	5✔
139	struct pollfd pfd;	2,181,793✔
140	memset(&pfd, 0, sizeof(pfd));	2,181,793✔
141	pfd.fd = socket;	2,181,793✔
142	if(rd)	2,181,793✔
143	pfd.events \|= POLLIN;	2,136,228✔
144	if(wr)	2,181,793✔
145	pfd.events \|= POLLOUT;	2,147,114✔
146	pfds.push_back(pfd);	2,181,793✔
147	}	2,181,793✔
148
149	int res = poll(&pfds[0], pfds.size(), (seconds < 0) ? -1 : (seconds * 1000)); // -1 = infinite	727,267✔
150	if(res < 0)	727,267!
151	unixDie("polling for activity from signers, "+std::to_string(d_sockets.size()));	×
152	pair<vector<int>, vector<int> > vects;	727,267✔
153	for(auto & pfd : pfds)	727,267✔
154	if(pfd.revents & POLLIN)	2,181,793✔
155	vects.first.push_back(pfd.fd);	270,086✔
156	else if(pfd.revents & POLLOUT)	1,911,707✔
157	vects.second.push_back(pfd.fd);	1,862,896✔
158
159	return vects;	727,267✔
160	}	727,267✔
161
162	void ChunkedSigningPipe::addSignedToChunks(std::unique_ptr<chunk_t>& signedChunk)
163	{	838,935✔
164	chunk_t::const_iterator from = signedChunk->begin();	838,935✔
165
166	while(from != signedChunk->end()) {	1,680,519✔
167	chunk_t& fillChunk = d_chunks.back();	841,584✔
168	chunk_t::size_type room = d_maxchunkrecords - fillChunk.size();	841,584✔
169
170	unsigned int fit = std::min(room, (chunk_t::size_type)(signedChunk->end() - from));	841,584✔
171
172	d_chunks.back().insert(fillChunk.end(), from , from + fit);	841,584✔
173	from+=fit;	841,584✔
174
175	if(from != signedChunk->end()) // it didn't fit, so add a new chunk	841,584✔
176	d_chunks.push_back(chunk_t());	3,682✔
177	}	841,584✔
178	}	838,935✔
179
180	void ChunkedSigningPipe::sendRRSetToWorker() // it sounds so socialist!
181	{	840,634✔
182	if(!d_mustSign) {	840,634✔
183	addSignedToChunks(d_rrsetToSign);	138,419✔
184	d_rrsetToSign->clear();	138,419✔
185	return;	138,419✔
186	}	138,419✔
187
188	if(d_final && !d_outstanding) // nothing to do!	702,215✔
189	return;	1,697✔
190
191	bool wantRead, wantWrite;	700,518✔
192
193	wantWrite = !d_rrsetToSign->empty();	700,518✔
194	wantRead = d_outstanding \|\| wantWrite; // if we wrote, we want to read	700,518!
195
196	pair<vector<int>, vector<int> > rwVect;	700,518✔
197
198	rwVect = waitForRW(wantRead, wantWrite, -1); // wait for something to happen	700,518✔
199
200	if(wantWrite && !rwVect.second.empty()) {	700,518✔
201	shuffle(rwVect.second.begin(), rwVect.second.end(), pdns::dns_random_engine()); // pick random available worker	685,327✔
202	auto ptr = d_rrsetToSign.get();	685,327✔
203	writen2(*rwVect.second.begin(), &ptr, sizeof(ptr));	685,327✔
204	// coverity[leaked_storage]
205	static_cast<void>(d_rrsetToSign.release());	685,327✔
206	d_rrsetToSign = make_unique<rrset_t>();	685,327✔
207	d_outstandings[*rwVect.second.begin()]++;	685,327✔
208	d_outstanding++;	685,327✔
209	d_queued++;	685,327✔
210	wantWrite=false;	685,327✔
211	}	685,327✔
212
213	if(wantRead) {	700,518✔
214	while(d_outstanding) {	712,078✔
215	for(int fd : rwVect.first) {	712,078✔
216	if(d_eof.count(fd))	270,086!
217	continue;	×
218
219	while(d_outstanding) {	970,603✔
220	chunk_t* chunk = nullptr;	965,204✔
221	int res = readn(fd, &chunk, sizeof(chunk));	965,204✔
222	if(!res) {	965,204✔
223	if (d_outstandings[fd] > 0) {	4!
224	throw std::runtime_error("A signing pipe worker died while we were waiting for its result");	×
225	}	×
226	d_eof.insert(fd);	4✔
227	break;	4✔
228	}	4✔
229	if(res < 0) {	965,200✔
230	if(errno != EAGAIN && errno != EINTR)	264,683!
231	unixDie("Error reading signed chunk from thread");	×
232	else	264,683✔
233	break;	264,683✔
234	}	264,683✔
235
236	std::unique_ptr<rrset_t> chunkPtr(chunk);	700,517✔
237	chunk = nullptr;	700,517✔
238	--d_outstanding;	700,517✔
239	d_outstandings[fd]--;	700,517✔
240
241	addSignedToChunks(chunkPtr);	700,517✔
242	}	700,517✔
243	}	270,086✔
244	if(!d_outstanding \|\| !d_final)	712,078✔
245	break;	700,518✔
246	rwVect = waitForRW(true, false, -1); // wait for something to happen	11,560✔
247	}	11,560✔
248	}	700,517✔
249
250	if(wantWrite) { // our optimization above failed, we now wait synchronously	700,518✔
251	rwVect = waitForRW(false, wantWrite, -1); // wait for something to happen	15,189✔
252	shuffle(rwVect.second.begin(), rwVect.second.end(), pdns::dns_random_engine()); // pick random available worker	15,189✔
253	auto ptr = d_rrsetToSign.get();	15,189✔
254	writen2(*rwVect.second.begin(), &ptr, sizeof(ptr));	15,189✔
255	// coverity[leaked_storage]
256	static_cast<void>(d_rrsetToSign.release());	15,189✔
257	d_rrsetToSign = make_unique<rrset_t>();	15,189✔
258	d_outstandings[*rwVect.second.begin()]++;	15,189✔
259	d_outstanding++;	15,189✔
260	d_queued++;	15,189✔
261	}	15,189✔
262
263	}	700,518✔
264
265	unsigned int ChunkedSigningPipe::getReady() const
266	{	×
267	unsigned int sum=0;	×
268	for(const auto& v : d_chunks) {	×
269	sum += v.size();	×
270	}	×
271	return sum;	×
272	}	×
273
274	void ChunkedSigningPipe::worker(int fd)
275	try	4,716✔
276	{	4,716✔
277	UeberBackend db("key-only");	4,716✔
278	DNSSECKeeper dk(&db);	4,716✔
279
280	chunk_t* chunk = nullptr;	4,716✔
281	int res;	4,716✔
282	for(;;) {	702,491✔
283	res = readn(fd, &chunk, sizeof(chunk));	702,491✔
284	if(!res)	702,491✔
285	break;	2,317✔
286	if(res < 0)	700,174!
287	unixDie("reading object pointer to sign from pdns");	×
288	try {	700,174✔
289	set<DNSName> authSet;	700,174✔
290	authSet.insert(d_signer);	700,174✔
291	addRRSigs(dk, db, authSet, *chunk);	700,174✔
292	++d_signed;	700,174✔
293
294	writen2(fd, &chunk, sizeof(chunk));	700,174✔
295	chunk = nullptr;	700,174✔
296	}	700,174✔
297	catch(const PDNSException& pe) {	700,174✔
298	delete chunk;	×
299	throw;	×
300	}	×
301	catch(const std::exception& e) {	700,174✔
302	delete chunk;	×
303	throw;	×
304	}	×
305	}	700,174✔
306	close(fd);	4,934✔
307	}	4,934✔
308	catch(const PDNSException& pe)	4,716✔
309	{	4,716✔
310	g_log<<Logger::Error<<"Signing thread died because of PDNSException: "<<pe.reason<<endl;	×
311	close(fd);	×
312	}	×
313	catch(const std::exception& e)	4,716✔
314	{	4,716✔
315	g_log<<Logger::Error<<"Signing thread died because of std::exception: "<<e.what()<<endl;	×
316	close(fd);	×
317	}	×
318
319	void ChunkedSigningPipe::flushToSign()
320	{	4,042✔
321	sendRRSetToWorker();	4,042✔
322	d_rrsetToSign->clear();	4,042✔
323	}	4,042✔
324
325	vector<DNSZoneRecord> ChunkedSigningPipe::getChunk(bool final)
326	{	35,163✔
327	if(final && !d_final) {	35,163✔
328	// this means we should keep on reading until d_outstanding == 0
329	d_final = true;	1,310✔
330	flushToSign();	1,310✔
331
332	for(int fd : d_sockets) {	2,362✔
333	shutdown(fd, SHUT_WR); // perhaps this transmits EOF the other side	2,358✔
334	//cerr<<"shutdown of "<<fd<<endl;
335	}	2,358✔
336	}	1,310✔
337	if(d_final)	35,163✔
338	flushToSign(); // should help us wait	2,732✔
339	vector<DNSZoneRecord> front=d_chunks.front();	35,163✔
340	d_chunks.pop_front();	35,163✔
341	if(d_chunks.empty())	35,163✔
342	d_chunks.push_back(vector<DNSZoneRecord>());	31,481✔
343	/* if(d_final && front.empty())
344	cerr<<"getChunk returning empty in final"<<endl; */
345	return front;	35,163✔
346	}	35,163✔
347
348

PowerDNS / pdns / 8372921153

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous