1/*  Part of SWI-Prolog
    2
    3    Author:        Jan Wielemaker
    4    E-mail:        J.Wielemaker@vu.nl
    5    WWW:           http://www.swi-prolog.org
    6    Copyright (c)  2006-2015, University of Amsterdam
    7                              VU University Amsterdam
    8    All rights reserved.
    9
   10    Redistribution and use in source and binary forms, with or without
   11    modification, are permitted provided that the following conditions
   12    are met:
   13
   14    1. Redistributions of source code must retain the above copyright
   15       notice, this list of conditions and the following disclaimer.
   16
   17    2. Redistributions in binary form must reproduce the above copyright
   18       notice, this list of conditions and the following disclaimer in
   19       the documentation and/or other materials provided with the
   20       distribution.
   21
   22    THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
   23    "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
   24    LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
   25    FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
   26    COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
   27    INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
   28    BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
   29    LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
   30    CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
   31    LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
   32    ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
   33    POSSIBILITY OF SUCH DAMAGE.
   34*/
   35
   36:- module(rdf_persistency,
   37          [ rdf_attach_db/2,            % +Directory, +Options
   38            rdf_detach_db/0,            % +Detach current Graph
   39            rdf_current_db/1,           % -Directory
   40            rdf_persistency/2,          % +Graph, +Bool
   41            rdf_flush_journals/1,       % +Options
   42            rdf_persistency_property/1, % ?Property
   43            rdf_journal_file/2,         % ?Graph, ?JournalFile
   44            rdf_snapshot_file/2,        % ?Graph, ?SnapshotFile
   45            rdf_db_to_file/2            % ?Graph, ?FileBase
   46          ]).   47:- use_module(library(semweb/rdf_db)).   48:- use_module(library(filesex)).   49:- use_module(library(lists)).   50:- use_module(library(uri)).   51:- use_module(library(debug)).   52:- use_module(library(option)).   53:- use_module(library(error)).   54:- use_module(library(thread)).   55:- use_module(library(apply)).   56
   57
   58/** <module> RDF persistency plugin
   59
   60This  module  provides  persistency   for    rdf_db.pl   based   on  the
   61rdf_monitor/2 predicate to  track  changes   to  the  repository.  Where
   62previous  versions  used  autosafe  of  the  whole  database  using  the
   63quick-load format of rdf_db, this version is  based on a quick-load file
   64per source (4th argument of rdf/4), and journalling for edit operations.
   65
   66The result is safe, avoids frequent small   changes to large files which
   67makes synchronisation and backup expensive and avoids long disruption of
   68the server doing the autosafe. Only loading large files disrupts service
   69for some time.
   70
   71The persistent backup of the database is  realised in a directory, using
   72a lock file to avoid corruption due to concurrent access. Each source is
   73represented by two files, the latest snapshot   and a journal. The state
   74is restored by loading  the  snapshot   and  replaying  the journal. The
   75predicate rdf_flush_journals/1 can be used to create fresh snapshots and
   76delete the journals.
   77
   78@tbd If there is a complete `.new'   snapshot  and no journal, we should
   79     move the .new to the plain snapshot name as a means of recovery.
   80
   81@tbd Backup of each graph using one or two files is very costly if there
   82     are many graphs.  Although the currently used subdirectories avoid
   83     hitting OS limits early, this is still not ideal. Probably we
   84     should collect (small, older?) files and combine them into a single
   85     quick load file.  We could call this (similar to GIT) a `pack'.
   86
   87@see    rdf_edit.pl
   88*/
   89
   90:- volatile
   91    rdf_directory/1,
   92    rdf_lock/2,
   93    rdf_option/1,
   94    source_journal_fd/2,
   95    file_base_db/2.   96:- dynamic
   97    rdf_directory/1,                % Absolute path
   98    rdf_lock/2,                     % Dir, Lock
   99    rdf_option/1,                   % Defined options
  100    source_journal_fd/2,            % DB, JournalFD
  101    file_base_db/2.                 % FileBase, DB
  102
  103:- meta_predicate
  104    no_agc(0).  105
  106:- predicate_options(rdf_attach_db/2, 2,
  107                     [ access(oneof([read_write,read_only])),
  108                       concurrency(positive_integer),
  109                       max_open_journals(positive_integer),
  110                       silent(oneof([true,false,brief])),
  111                       log_nested_transactions(boolean)
  112                     ]).  113
  114%!  rdf_attach_db(+Directory, +Options) is det.
  115%
  116%   Start persistent operations using Directory   as  place to store
  117%   files.   There are several cases:
  118%
  119%           * Empty DB, existing directory
  120%           Load the DB from the existing directory
  121%
  122%           * Full DB, empty directory
  123%           Create snapshots for all sources in directory
  124%
  125%   Options:
  126%
  127%           * access(+AccessMode)
  128%           One of =auto= (default), =read_write= or
  129%           =read_only=. Read-only access implies that the RDF
  130%           store is not locked. It is read at startup and all
  131%           modifications to the data are temporary. The default
  132%           =auto= mode is =read_write= if the directory is
  133%           writeable and the lock can be acquired.  Otherwise
  134%           it reverts to =read_only=.
  135%
  136%           * concurrency(+Jobs)
  137%           Number of threads to use for loading the initial
  138%           database.  If not provided it is the number of CPUs
  139%           as optained from the flag =cpu_count=.
  140%
  141%           * max_open_journals(+Count)
  142%           Maximum number of journals kept open.  If not provided,
  143%           the default is 10.  See limit_fd_pool/0.
  144%
  145%           * directory_levels(+Count)
  146%           Number of levels of intermediate directories for storing
  147%           the graph files.  Default is 2.
  148%
  149%           * silent(+BoolOrBrief)
  150%           If =true= (default =false=), do not print informational
  151%           messages.  Finally, if =brief= it will show minimal
  152%           feedback.
  153%
  154%           * log_nested_transactions(+Boolean)
  155%           If =true=, nested _log_ transactions are added to the
  156%           journal information.  By default (=false=), no log-term
  157%           is added for nested transactions.\\
  158%
  159%   @error existence_error(source_sink, Directory)
  160%   @error permission_error(write, directory, Directory)
  161
  162rdf_attach_db(DirSpec, Options) :-
  163    option(access(read_only), Options),
  164    !,
  165    absolute_file_name(DirSpec,
  166                       Directory,
  167                       [ access(read),
  168                         file_type(directory)
  169                       ]),
  170    rdf_attach_db_ro(Directory, Options).
  171rdf_attach_db(DirSpec, Options) :-
  172    option(access(read_write), Options),
  173    !,
  174    rdf_attach_db_rw(DirSpec, Options).
  175rdf_attach_db(DirSpec, Options) :-
  176    absolute_file_name(DirSpec,
  177                       Directory,
  178                       [ access(exist),
  179                         file_type(directory),
  180                         file_errors(fail)
  181                       ]),
  182    !,
  183    (   access_file(Directory, write)
  184    ->  catch(rdf_attach_db_rw(Directory, Options), E, true),
  185        (   var(E)
  186        ->  true
  187        ;   E = error(permission_error(lock, rdf_db, _), _)
  188        ->  print_message(warning, E),
  189            print_message(warning, rdf(read_only)),
  190            rdf_attach_db(DirSpec, [access(read_only)|Options])
  191        ;   throw(E)
  192        )
  193    ;   print_message(warning,
  194                      error(permission_error(write, directory, Directory))),
  195        print_message(warning, rdf(read_only)),
  196        rdf_attach_db_ro(Directory, Options)
  197    ).
  198rdf_attach_db(DirSpec, Options) :-
  199    catch(rdf_attach_db_rw(DirSpec, Options), E, true),
  200    (   var(E)
  201    ->  true
  202    ;   print_message(warning, E),
  203        print_message(warning, rdf(read_only)),
  204        rdf_attach_db(DirSpec, [access(read_only)|Options])
  205    ).
  206
  207
  208rdf_attach_db_rw(DirSpec, Options) :-
  209    absolute_file_name(DirSpec,
  210                       Directory,
  211                       [ access(write),
  212                         file_type(directory),
  213                         file_errors(fail)
  214                       ]),
  215    !,
  216    (   rdf_directory(Directory)
  217    ->  true                        % update settings?
  218    ;   rdf_detach_db,
  219        mkdir(Directory),
  220        lock_db(Directory),
  221        assert(rdf_directory(Directory)),
  222        assert_options(Options),
  223        stop_monitor,               % make sure not to register load
  224        no_agc(load_db),
  225        at_halt(rdf_detach_db),
  226        start_monitor
  227    ).
  228rdf_attach_db_rw(DirSpec, Options) :-
  229    absolute_file_name(DirSpec,
  230                       Directory,
  231                       [ solutions(all)
  232                       ]),
  233    (   exists_directory(Directory)
  234    ->  access_file(Directory, write)
  235    ;   catch(make_directory(Directory), _, fail)
  236    ),
  237    !,
  238    rdf_attach_db(Directory, Options).
  239rdf_attach_db_rw(DirSpec, _) :-         % Generate an existence or
  240    absolute_file_name(DirSpec,     % permission error
  241                       Directory,
  242                       [ access(exist),
  243                         file_type(directory)
  244                       ]),
  245    permission_error(write, directory, Directory).
  246
  247%!  rdf_attach_db_ro(+Directory, +Options)
  248%
  249%   Open an RDF database in read-only mode.
  250
  251rdf_attach_db_ro(Directory, Options) :-
  252    rdf_detach_db,
  253    assert(rdf_directory(Directory)),
  254    assert_options(Options),
  255    stop_monitor,           % make sure not to register load
  256    no_agc(load_db).
  257
  258
  259assert_options([]).
  260assert_options([H|T]) :-
  261    (   option_type(H, Check)
  262    ->  Check,
  263        assert(rdf_option(H))
  264    ;   true                        % ignore options we do not understand
  265    ),
  266    assert_options(T).
  267
  268option_type(concurrency(X),             must_be(positive_integer, X)).
  269option_type(max_open_journals(X),       must_be(positive_integer, X)).
  270option_type(directory_levels(X),        must_be(positive_integer, X)).
  271option_type(silent(X),                  must_be(oneof([true,false,brief]), X)).
  272option_type(log_nested_transactions(X), must_be(boolean, X)).
  273option_type(access(X),                  must_be(oneof([read_write,
  274                                                       read_only]), X)).
  275
  276
  277%!  rdf_persistency_property(?Property) is nondet.
  278%
  279%   True if Property  is  a  property   of  the  current  persistent
  280%   database. Currently makes to options   passed to rdf_attach_db/2
  281%   available.  Notable  rdf_persistency_property(access(read_only))
  282%   is true if the database  is   mounted  in  read-only mode. Other
  283%   properties:
  284%
  285%     - directory(Dir)
  286%     Directory in which the database resides.
  287
  288rdf_persistency_property(Property) :-
  289    var(Property),
  290    !,
  291    rdf_persistency_property_(Property).
  292rdf_persistency_property(Property) :-
  293    rdf_persistency_property_(Property),
  294    !.
  295
  296rdf_persistency_property_(Property) :-
  297    rdf_option(Property).
  298rdf_persistency_property_(directory(Dir)) :-
  299    rdf_directory(Dir).
  300
  301%!  no_agc(:Goal)
  302%
  303%   Run Goal with atom garbage collection   disabled. Loading an RDF
  304%   database creates large amounts  of  atoms   we  *know*  are  not
  305%   garbage.
  306
  307no_agc(Goal) :-
  308    current_prolog_flag(agc_margin, Old),
  309    setup_call_cleanup(
  310        set_prolog_flag(agc_margin, 0),
  311        Goal,
  312        set_prolog_flag(agc_margin, Old)).
  313
  314
  315%!  rdf_detach_db is det.
  316%
  317%   Detach from the  current  database.   Succeeds  silently  if  no
  318%   database is attached. Normally called at  the end of the program
  319%   through at_halt/1.
  320
  321rdf_detach_db :-
  322    debug(halt, 'Detaching RDF database', []),
  323    stop_monitor,
  324    close_journals,
  325    (   retract(rdf_directory(Dir))
  326    ->  debug(halt, 'DB Directory: ~w', [Dir]),
  327        save_prefixes(Dir),
  328        retractall(rdf_option(_)),
  329        retractall(source_journal_fd(_,_)),
  330        unlock_db(Dir)
  331    ;   true
  332    ).
  333
  334
  335%!  rdf_current_db(?Dir)
  336%
  337%   True if Dir is the current RDF persistent database.
  338
  339rdf_current_db(Directory) :-
  340    rdf_directory(Dir),
  341    !,
  342    Dir = Directory.
  343
  344
  345%!  rdf_flush_journals(+Options)
  346%
  347%   Flush dirty journals.  Options:
  348%
  349%           * min_size(+KB)
  350%           Only flush if journal is over KB in size.
  351%           * graph(+Graph)
  352%           Only flush the journal of Graph
  353%
  354%   @tbd Provide a default for min_size?
  355
  356rdf_flush_journals(Options) :-
  357    option(graph(Graph), Options, _),
  358    forall(rdf_graph(Graph),
  359           rdf_flush_journal(Graph, Options)).
  360
  361rdf_flush_journal(Graph, Options) :-
  362    db_files(Graph, _SnapshotFile, JournalFile),
  363    db_file(JournalFile, File),
  364    (   \+ exists_file(File)
  365    ->  true
  366    ;   memberchk(min_size(KB), Options),
  367        size_file(JournalFile, Size),
  368        Size / 1024 < KB
  369    ->  true
  370    ;   create_db(Graph)
  371    ).
  372
  373                 /*******************************
  374                 *             LOAD             *
  375                 *******************************/
  376
  377%!  load_db is det.
  378%
  379%   Reload database from the directory specified by rdf_directory/1.
  380%   First we find all names graphs using find_dbs/1 and then we load
  381%   them.
  382
  383load_db :-
  384    rdf_directory(Dir),
  385    concurrency(Jobs),
  386    cpu_stat_key(Jobs, StatKey),
  387    get_time(Wall0),
  388    statistics(StatKey, T0),
  389    load_prefixes(Dir),
  390    verbosity(Silent),
  391    find_dbs(Dir, Graphs, SnapShots, Journals),
  392    length(Graphs, GraphCount),
  393    maplist(rdf_unload_graph, Graphs),
  394    rdf_statistics(triples(Triples0)),
  395    load_sources(snapshots, SnapShots, Silent, Jobs),
  396    load_sources(journals, Journals, Silent, Jobs),
  397    rdf_statistics(triples(Triples1)),
  398    statistics(StatKey, T1),
  399    get_time(Wall1),
  400    T is T1 - T0,
  401    Wall is Wall1 - Wall0,
  402    Triples = Triples1 - Triples0,
  403    message_level(Silent, Level),
  404    print_message(Level, rdf(restore(attached(GraphCount, Triples, T/Wall)))).
  405
  406load_sources(_, [], _, _) :- !.
  407load_sources(Type, Sources, Silent, Jobs) :-
  408    length(Sources, Count),
  409    RunJobs is min(Count, Jobs),
  410    print_message(informational, rdf(restoring(Type, Count, RunJobs))),
  411    make_goals(Sources, Silent, 1, Count, Goals),
  412    concurrent(RunJobs, Goals, []).
  413
  414
  415%!  make_goals(+DBs, +Silent, +Index, +Total, -Goals)
  416
  417make_goals([], _, _, _, []).
  418make_goals([DB|T0], Silent, I,  Total,
  419           [load_source(DB, Silent, I, Total)|T]) :-
  420    I2 is I + 1,
  421    make_goals(T0, Silent, I2, Total, T).
  422
  423verbosity(Silent) :-
  424    rdf_option(silent(Silent)),
  425    !.
  426verbosity(Silent) :-
  427    current_prolog_flag(verbose, silent),
  428    !,
  429    Silent = true.
  430verbosity(brief).
  431
  432
  433%!  concurrency(-Jobs)
  434%
  435%   Number of jobs to run concurrently.
  436
  437concurrency(Jobs) :-
  438    rdf_option(concurrency(Jobs)),
  439    !.
  440concurrency(Jobs) :-
  441    current_prolog_flag(cpu_count, Jobs),
  442    Jobs > 0,
  443    !.
  444concurrency(1).
  445
  446cpu_stat_key(1, cputime) :- !.
  447cpu_stat_key(_, process_cputime).
  448
  449
  450%!  find_dbs(+Dir, -Graphs, -SnapBySize, -JournalBySize) is det.
  451%
  452%   Scan the persistent database and return a list of snapshots and
  453%   journals, both sorted by file-size.  Each term is of the form
  454%
  455%     ==
  456%     db(Size, Ext, DB, DBFile, Depth)
  457%     ==
  458
  459find_dbs(Dir, Graphs, SnapBySize, JournalBySize) :-
  460    directory_files(Dir, Files),
  461    phrase(scan_db_files(Files, Dir, '.', 0), Scanned),
  462    maplist(db_graph, Scanned, UnsortedGraphs),
  463    sort(UnsortedGraphs, Graphs),
  464    (   consider_reindex_db(Dir, Graphs, Scanned)
  465    ->  find_dbs(Dir, Graphs, SnapBySize, JournalBySize)
  466    ;   partition(db_is_snapshot, Scanned, Snapshots, Journals),
  467        sort(Snapshots, SnapBySize),
  468        sort(Journals, JournalBySize)
  469    ).
  470
  471consider_reindex_db(Dir, Graphs, Scanned) :-
  472    length(Graphs, Count),
  473    Count > 0,
  474    DepthNeeded is floor(log(Count)/log(256)),
  475    (   maplist(depth_db(DepthNow), Scanned)
  476    ->  (   DepthNeeded > DepthNow
  477        ->  true
  478        ;   retractall(rdf_option(directory_levels(_))),
  479            assertz(rdf_option(directory_levels(DepthNow))),
  480            fail
  481        )
  482    ;   true
  483    ),
  484    reindex_db(Dir, DepthNeeded).
  485
  486db_is_snapshot(Term) :-
  487    arg(2, Term, trp).
  488
  489db_graph(Term, DB) :-
  490    arg(3, Term, DB).
  491
  492db_file_name(Term, File) :-
  493    arg(4, Term, File).
  494
  495depth_db(Depth, DB) :-
  496    arg(5, DB, Depth).
  497
  498%!  scan_db_files(+Files, +Dir, +Prefix, +Depth)// is det.
  499%
  500%   Produces a list of db(DB,  Size,   File)  for all recognised RDF
  501%   database files.  File is relative to the database directory Dir.
  502
  503scan_db_files([], _, _, _) -->
  504    [].
  505scan_db_files([Nofollow|T], Dir, Prefix, Depth) -->
  506    { nofollow(Nofollow) },
  507    !,
  508    scan_db_files(T, Dir, Prefix, Depth).
  509scan_db_files([File|T], Dir, Prefix, Depth) -->
  510    { file_name_extension(Base, Ext, File),
  511      db_extension(Ext),
  512      !,
  513      rdf_db_to_file(DB, Base),
  514      directory_file_path(Prefix, File, DBFile),
  515      directory_file_path(Dir, DBFile, AbsFile),
  516      size_file(AbsFile, Size)
  517    },
  518    [ db(Size, Ext, DB, AbsFile, Depth) ],
  519    scan_db_files(T, Dir, Prefix, Depth).
  520scan_db_files([D|T], Dir, Prefix, Depth) -->
  521    { directory_file_path(Prefix, D, SubD),
  522      directory_file_path(Dir, SubD, AbsD),
  523      exists_directory(AbsD),
  524      \+ read_link(AbsD, _, _),    % Do not follow links
  525      !,
  526      directory_files(AbsD, SubFiles),
  527      SubDepth is Depth + 1
  528    },
  529    scan_db_files(SubFiles, Dir, SubD, SubDepth),
  530    scan_db_files(T, Dir, Prefix, Depth).
  531scan_db_files([_|T], Dir, Prefix, Depth) -->
  532    scan_db_files(T, Dir, Prefix, Depth).
  533
  534nofollow(.).
  535nofollow(..).
  536
  537db_extension(trp).
  538db_extension(jrn).
  539
  540:- public load_source/4.                % called through make_goals/5
  541
  542load_source(DB, Silent, Nth, Total) :-
  543    db_file_name(DB, File),
  544    db_graph(DB, Graph),
  545    message_level(Silent, Level),
  546    graph_triple_count(Graph, Count0),
  547    statistics(cputime, T0),
  548    (   db_is_snapshot(DB)
  549    ->  print_message(Level, rdf(restore(Silent, snapshot(Graph, File)))),
  550        rdf_load_db(File)
  551    ;   print_message(Level, rdf(restore(Silent, journal(Graph, File)))),
  552        load_journal(File, Graph)
  553    ),
  554    statistics(cputime, T1),
  555    T is T1 - T0,
  556    graph_triple_count(Graph, Count1),
  557    Count is Count1 - Count0,
  558    print_message(Level, rdf(restore(Silent,
  559                                     done(Graph, T, Count, Nth, Total)))).
  560
  561
  562graph_triple_count(Graph, Count) :-
  563    rdf_statistics(triples_by_graph(Graph, Count)),
  564    !.
  565graph_triple_count(_, 0).
  566
  567
  568%!  attach_graph(+Graph, +Options) is det.
  569%
  570%   Load triples and reload  journal   from  the  indicated snapshot
  571%   file.
  572
  573attach_graph(Graph, Options) :-
  574    (   option(silent(true), Options)
  575    ->  Level = silent
  576    ;   Level = informational
  577    ),
  578    db_files(Graph, SnapshotFile, JournalFile),
  579    rdf_retractall(_,_,_,Graph),
  580    statistics(cputime, T0),
  581    print_message(Level, rdf(restore(Silent, Graph))),
  582    db_file(SnapshotFile, AbsSnapShot),
  583    (   exists_file(AbsSnapShot)
  584    ->  print_message(Level, rdf(restore(Silent, snapshot(SnapshotFile)))),
  585        rdf_load_db(AbsSnapShot)
  586    ;   true
  587    ),
  588    (   exists_db(JournalFile)
  589    ->  print_message(Level, rdf(restore(Silent, journal(JournalFile)))),
  590        load_journal(JournalFile, Graph)
  591    ;   true
  592    ),
  593    statistics(cputime, T1),
  594    T is T1 - T0,
  595    (   rdf_statistics(triples_by_graph(Graph, Count))
  596    ->  true
  597    ;   Count = 0
  598    ),
  599    print_message(Level, rdf(restore(Silent,
  600                                     done(Graph, T, Count)))).
  601
  602message_level(true, silent) :- !.
  603message_level(_, informational).
  604
  605
  606                 /*******************************
  607                 *         LOAD JOURNAL         *
  608                 *******************************/
  609
  610%!  load_journal(+File:atom, +DB:atom) is det.
  611%
  612%   Process transactions from the RDF journal File, adding the given
  613%   named graph.
  614
  615load_journal(File, DB) :-
  616    rdf_create_graph(DB),
  617    setup_call_cleanup(
  618        open(File, read, In, [encoding(utf8)]),
  619        ( read(In, T0),
  620          process_journal(T0, In, DB)
  621        ),
  622        close(In)).
  623
  624process_journal(end_of_file, _, _) :- !.
  625process_journal(Term, In, DB) :-
  626    (   process_journal_term(Term, DB)
  627    ->  true
  628    ;   throw(error(type_error(journal_term, Term), _))
  629    ),
  630    read(In, T2),
  631    process_journal(T2, In, DB).
  632
  633process_journal_term(assert(S,P,O), DB) :-
  634    rdf_assert(S,P,O,DB).
  635process_journal_term(assert(S,P,O,Line), DB) :-
  636    rdf_assert(S,P,O,DB:Line).
  637process_journal_term(retract(S,P,O), DB) :-
  638    rdf_retractall(S,P,O,DB).
  639process_journal_term(retract(S,P,O,Line), DB) :-
  640    rdf_retractall(S,P,O,DB:Line).
  641process_journal_term(update(S,P,O,Action), DB) :-
  642    (   rdf_update(S,P,O,DB, Action)
  643    ->  true
  644    ;   print_message(warning, rdf(update_failed(S,P,O,Action)))
  645    ).
  646process_journal_term(start(_), _).      % journal open/close
  647process_journal_term(end(_), _).
  648process_journal_term(begin(_), _).      % logged transaction (compatibility)
  649process_journal_term(end, _).
  650process_journal_term(begin(_,_,_,_), _). % logged transaction (current)
  651process_journal_term(end(_,_,_), _).
  652
  653
  654                 /*******************************
  655                 *         CREATE JOURNAL       *
  656                 *******************************/
  657
  658:- dynamic
  659    blocked_db/2,                   % DB, Reason
  660    transaction_message/3,          % Nesting, Time, Message
  661    transaction_db/3.               % Nesting, DB, Id
  662
  663%!  rdf_persistency(+DB, Bool)
  664%
  665%   Specify whether a database is persistent.  Switching to =false=
  666%   kills the persistent state.  Switching to =true= creates it.
  667
  668rdf_persistency(DB, Bool) :-
  669    must_be(atom, DB),
  670    must_be(boolean, Bool),
  671    fail.
  672rdf_persistency(DB, false) :-
  673    !,
  674    (   blocked_db(DB, persistency)
  675    ->  true
  676    ;   assert(blocked_db(DB, persistency)),
  677        delete_db(DB)
  678    ).
  679rdf_persistency(DB, true) :-
  680    (   retract(blocked_db(DB, persistency))
  681    ->  create_db(DB)
  682    ;   true
  683    ).
  684
  685%!  rdf_db:property_of_graph(?Property, +Graph) is nondet.
  686%
  687%   Extend rdf_graph_property/2 with new properties.
  688
  689:- multifile
  690    rdf_db:property_of_graph/2.  691
  692rdf_db:property_of_graph(persistent(State), Graph) :-
  693    (   blocked_db(Graph, persistency)
  694    ->  State = false
  695    ;   State = true
  696    ).
  697
  698
  699%!  start_monitor is det.
  700%!  stop_monitor is det.
  701%
  702%   Start/stop monitoring the RDF database   for  changes and update
  703%   the journal.
  704
  705start_monitor :-
  706    rdf_monitor(monitor,
  707                [ -assert(load)
  708                ]).
  709stop_monitor :-
  710    rdf_monitor(monitor,
  711                [ -all
  712                ]).
  713
  714%!  monitor(+Term) is semidet.
  715%
  716%   Handle an rdf_monitor/2 callback to  deal with persistency. Note
  717%   that the monitor calls that come   from rdf_db.pl that deal with
  718%   database changes are serialized.  They   do  come from different
  719%   threads though.
  720
  721monitor(Msg) :-
  722    debug(monitor, 'Monitor: ~p~n', [Msg]),
  723    fail.
  724monitor(assert(S,P,O,DB:Line)) :-
  725    !,
  726    \+ blocked_db(DB, _),
  727    journal_fd(DB, Fd),
  728    open_transaction(DB, Fd),
  729    format(Fd, '~q.~n', [assert(S,P,O,Line)]),
  730    sync_journal(DB, Fd).
  731monitor(assert(S,P,O,DB)) :-
  732    \+ blocked_db(DB, _),
  733    journal_fd(DB, Fd),
  734    open_transaction(DB, Fd),
  735    format(Fd, '~q.~n', [assert(S,P,O)]),
  736    sync_journal(DB, Fd).
  737monitor(retract(S,P,O,DB:Line)) :-
  738    !,
  739    \+ blocked_db(DB, _),
  740    journal_fd(DB, Fd),
  741    open_transaction(DB, Fd),
  742    format(Fd, '~q.~n', [retract(S,P,O,Line)]),
  743    sync_journal(DB, Fd).
  744monitor(retract(S,P,O,DB)) :-
  745    \+ blocked_db(DB, _),
  746    journal_fd(DB, Fd),
  747    open_transaction(DB, Fd),
  748    format(Fd, '~q.~n', [retract(S,P,O)]),
  749    sync_journal(DB, Fd).
  750monitor(update(S,P,O,DB:Line,Action)) :-
  751    !,
  752    \+ blocked_db(DB, _),
  753    (   Action = graph(NewDB)
  754    ->  monitor(assert(S,P,O,NewDB)),
  755        monitor(retract(S,P,O,DB:Line))
  756    ;   journal_fd(DB, Fd),
  757        format(Fd, '~q.~n', [update(S,P,O,Action)]),
  758        sync_journal(DB, Fd)
  759    ).
  760monitor(update(S,P,O,DB,Action)) :-
  761    \+ blocked_db(DB, _),
  762    (   Action = graph(NewDB)
  763    ->  monitor(assert(S,P,O,NewDB)),
  764        monitor(retract(S,P,O,DB))
  765    ;   journal_fd(DB, Fd),
  766        open_transaction(DB, Fd),
  767        format(Fd, '~q.~n', [update(S,P,O,Action)]),
  768        sync_journal(DB, Fd)
  769    ).
  770monitor(load(BE, _DumpFileURI)) :-
  771    (   BE = end(Graphs)
  772    ->  sync_loaded_graphs(Graphs)
  773    ;   true
  774    ).
  775monitor(create_graph(Graph)) :-
  776    \+ blocked_db(Graph, _),
  777    journal_fd(Graph, Fd),
  778    open_transaction(Graph, Fd),
  779    sync_journal(Graph, Fd).
  780monitor(reset) :-
  781    forall(rdf_graph(Graph), delete_db(Graph)).
  782                                        % TBD: Remove empty directories?
  783
  784monitor(transaction(BE, Id)) :-
  785    monitor_transaction(Id, BE).
  786
  787monitor_transaction(load_journal(DB), begin(_)) :-
  788    !,
  789    assert(blocked_db(DB, journal)).
  790monitor_transaction(load_journal(DB), end(_)) :-
  791    !,
  792    retractall(blocked_db(DB, journal)).
  793
  794monitor_transaction(parse(URI), begin(_)) :-
  795    !,
  796    (   blocked_db(URI, persistency)
  797    ->  true
  798    ;   assert(blocked_db(URI, parse))
  799    ).
  800monitor_transaction(parse(URI), end(_)) :-
  801    !,
  802    (   retract(blocked_db(URI, parse))
  803    ->  create_db(URI)
  804    ;   true
  805    ).
  806monitor_transaction(unload(DB), begin(_)) :-
  807    !,
  808    (   blocked_db(DB, persistency)
  809    ->  true
  810    ;   assert(blocked_db(DB, unload))
  811    ).
  812monitor_transaction(unload(DB), end(_)) :-
  813    !,
  814    (   retract(blocked_db(DB, unload))
  815    ->  delete_db(DB)
  816    ;   true
  817    ).
  818monitor_transaction(log(Msg), begin(N)) :-
  819    !,
  820    check_nested(N),
  821    get_time(Time),
  822    asserta(transaction_message(N, Time, Msg)).
  823monitor_transaction(log(_), end(N)) :-
  824    check_nested(N),
  825    retract(transaction_message(N, _, _)),
  826    !,
  827    findall(DB:Id, retract(transaction_db(N, DB, Id)), DBs),
  828    end_transactions(DBs, N).
  829monitor_transaction(log(Msg, DB), begin(N)) :-
  830    !,
  831    check_nested(N),
  832    get_time(Time),
  833    asserta(transaction_message(N, Time, Msg)),
  834    journal_fd(DB, Fd),
  835    open_transaction(DB, Fd).
  836monitor_transaction(log(Msg, _DB), end(N)) :-
  837    monitor_transaction(log(Msg), end(N)).
  838
  839
  840%!  check_nested(+Level) is semidet.
  841%
  842%   True if we must log this transaction.   This  is always the case
  843%   for toplevel transactions. Nested transactions   are only logged
  844%   if log_nested_transactions(true) is defined.
  845
  846check_nested(0) :- !.
  847check_nested(_) :-
  848    rdf_option(log_nested_transactions(true)).
  849
  850
  851%!  open_transaction(+DB, +Fd) is det.
  852%
  853%   Add a begin(Id, Level, Time,  Message)   term  if  a transaction
  854%   involves DB. Id is an incremental   integer, where each database
  855%   has its own counter. Level is the nesting level, Time a floating
  856%   point timestamp and Message te message   provided as argument to
  857%   the log message.
  858
  859open_transaction(DB, Fd) :-
  860    transaction_message(N, Time, Msg),
  861    !,
  862    (   transaction_db(N, DB, _)
  863    ->  true
  864    ;   next_transaction_id(DB, Id),
  865        assert(transaction_db(N, DB, Id)),
  866        RoundedTime is round(Time*100)/100,
  867        format(Fd, '~q.~n', [begin(Id, N, RoundedTime, Msg)])
  868    ).
  869open_transaction(_,_).
  870
  871
  872%!  next_transaction_id(+DB, -Id) is det.
  873%
  874%   Id is the number to user for  the next logged transaction on DB.
  875%   Transactions in each  named  graph   are  numbered  in sequence.
  876%   Searching the Id of the last transaction is performed by the 2nd
  877%   clause starting 1Kb from the end   and doubling this offset each
  878%   failure.
  879
  880:- dynamic
  881    current_transaction_id/2.  882
  883next_transaction_id(DB, Id) :-
  884    retract(current_transaction_id(DB, Last)),
  885    !,
  886    Id is Last + 1,
  887    assert(current_transaction_id(DB, Id)).
  888next_transaction_id(DB, Id) :-
  889    db_files(DB, _, Journal),
  890    exists_file(Journal),
  891    !,
  892    size_file(Journal, Size),
  893    open_db(Journal, read, In, []),
  894    call_cleanup(iterative_expand(In, Size, Last), close(In)),
  895    Id is Last + 1,
  896    assert(current_transaction_id(DB, Id)).
  897next_transaction_id(DB, 1) :-
  898    assert(current_transaction_id(DB, 1)).
  899
  900iterative_expand(_, 0, 0) :- !.
  901iterative_expand(In, Size, Last) :-     % Scan growing sections from the end
  902    Max is floor(log(Size)/log(2)),
  903    between(10, Max, Step),
  904    Offset is -(1<<Step),
  905    seek(In, Offset, eof, _),
  906    skip(In, 10),                   % records are line-based
  907    read(In, T0),
  908    last_transaction_id(T0, In, 0, Last),
  909    Last > 0,
  910    !.
  911iterative_expand(In, _, Last) :-        % Scan the whole file
  912    seek(In, 0, bof, _),
  913    read(In, T0),
  914    last_transaction_id(T0, In, 0, Last).
  915
  916last_transaction_id(end_of_file, _, Last, Last) :- !.
  917last_transaction_id(end(Id, _, _), In, _, Last) :-
  918    read(In, T1),
  919    last_transaction_id(T1, In, Id, Last).
  920last_transaction_id(_, In, Id, Last) :-
  921    read(In, T1),
  922    last_transaction_id(T1, In, Id, Last).
  923
  924
  925%!  end_transactions(+DBs:list(atom:id)) is det.
  926%
  927%   End a transaction that affected the  given list of databases. We
  928%   write the list of other affected databases as an argument to the
  929%   end-term to facilitate fast finding of the related transactions.
  930%
  931%   In each database, the transaction is   ended with a term end(Id,
  932%   Nesting, Others), where  Id  and   Nesting  are  the transaction
  933%   identifier and nesting (see open_transaction/2)  and Others is a
  934%   list of DB:Id,  indicating  other   databases  affected  by  the
  935%   transaction.
  936
  937end_transactions(DBs, N) :-
  938    end_transactions(DBs, DBs, N).
  939
  940end_transactions([], _, _).
  941end_transactions([DB:Id|T], DBs, N) :-
  942    journal_fd(DB, Fd),
  943    once(select(DB:Id, DBs, Others)),
  944    format(Fd, 'end(~q, ~q, ~q).~n', [Id, N, Others]),
  945    sync_journal(DB, Fd),
  946    end_transactions(T, DBs, N).
  947
  948
  949%!  sync_loaded_graphs(+Graphs)
  950%
  951%   Called after a binary triple has been loaded that added triples
  952%   to the given graphs.
  953
  954sync_loaded_graphs(Graphs) :-
  955    maplist(create_db, Graphs).
  956
  957
  958                 /*******************************
  959                 *         JOURNAL FILES        *
  960                 *******************************/
  961
  962%!  journal_fd(+DB, -Stream) is det.
  963%
  964%   Get an open stream to a journal. If the journal is not open, old
  965%   journals are closed to satisfy   the =max_open_journals= option.
  966%   Then the journal is opened in   =append= mode. Journal files are
  967%   always encoded as UTF-8 for  portability   as  well as to ensure
  968%   full coverage of Unicode.
  969
  970journal_fd(DB, Fd) :-
  971    source_journal_fd(DB, Fd),
  972    !.
  973journal_fd(DB, Fd) :-
  974    with_mutex(rdf_journal_file,
  975               journal_fd_(DB, Out)),
  976    Fd = Out.
  977
  978journal_fd_(DB, Fd) :-
  979    source_journal_fd(DB, Fd),
  980    !.
  981journal_fd_(DB, Fd) :-
  982    limit_fd_pool,
  983    db_files(DB, _Snapshot, Journal),
  984    open_db(Journal, append, Fd,
  985            [ close_on_abort(false)
  986            ]),
  987    time_stamp(Now),
  988    format(Fd, '~q.~n', [start([time(Now)])]),
  989    assert(source_journal_fd(DB, Fd)).              % new one at the end
  990
  991%!  limit_fd_pool is det.
  992%
  993%   Limit the number of  open   journals  to max_open_journals (10).
  994%   Note that calls  from  rdf_monitor/2   are  issued  in different
  995%   threads, but as they are part of write operations they are fully
  996%   synchronised.
  997
  998limit_fd_pool :-
  999    predicate_property(source_journal_fd(_, _), number_of_clauses(N)),
 1000    !,
 1001    (   rdf_option(max_open_journals(Max))
 1002    ->  true
 1003    ;   Max = 10
 1004    ),
 1005    Close is N - Max,
 1006    forall(between(1, Close, _),
 1007           close_oldest_journal).
 1008limit_fd_pool.
 1009
 1010close_oldest_journal :-
 1011    source_journal_fd(DB, _Fd),
 1012    !,
 1013    debug(rdf_persistency, 'Closing old journal for ~q', [DB]),
 1014    close_journal(DB).
 1015close_oldest_journal.
 1016
 1017
 1018%!  sync_journal(+DB, +Fd)
 1019%
 1020%   Sync journal represented by database and   stream.  If the DB is
 1021%   involved in a transaction there is   no point flushing until the
 1022%   end of the transaction.
 1023
 1024sync_journal(DB, _) :-
 1025    transaction_db(_, DB, _),
 1026    !.
 1027sync_journal(_, Fd) :-
 1028    flush_output(Fd).
 1029
 1030%!  close_journal(+DB) is det.
 1031%
 1032%   Close the journal associated with DB if it is open.
 1033
 1034close_journal(DB) :-
 1035    with_mutex(rdf_journal_file,
 1036               close_journal_(DB)).
 1037
 1038close_journal_(DB) :-
 1039    (   retract(source_journal_fd(DB, Fd))
 1040    ->  time_stamp(Now),
 1041        format(Fd, '~q.~n', [end([time(Now)])]),
 1042        close(Fd, [force(true)])
 1043    ;   true
 1044    ).
 1045
 1046%!  close_journals
 1047%
 1048%   Close all open journals.
 1049
 1050close_journals :-
 1051    forall(source_journal_fd(DB, _),
 1052           catch(close_journal(DB), E,
 1053                 print_message(error, E))).
 1054
 1055%!  create_db(+Graph)
 1056%
 1057%   Create a saved version of Graph in corresponding file, close and
 1058%   delete journals.
 1059
 1060create_db(Graph) :-
 1061    \+ rdf(_,_,_,Graph),
 1062    !,
 1063    debug(rdf_persistency, 'Deleting empty Graph ~w', [Graph]),
 1064    delete_db(Graph).
 1065create_db(Graph) :-
 1066    debug(rdf_persistency, 'Saving Graph ~w', [Graph]),
 1067    close_journal(Graph),
 1068    db_abs_files(Graph, Snapshot, Journal),
 1069    atom_concat(Snapshot, '.new', NewSnapshot),
 1070    (   catch(( create_directory_levels(Snapshot),
 1071                rdf_save_db(NewSnapshot, Graph)
 1072              ), Error,
 1073              ( print_message(warning, Error),
 1074                fail
 1075              ))
 1076    ->  (   exists_file(Journal)
 1077        ->  delete_file(Journal)
 1078        ;   true
 1079        ),
 1080        rename_file(NewSnapshot, Snapshot),
 1081        debug(rdf_persistency, 'Saved Graph ~w', [Graph])
 1082    ;   catch(delete_file(NewSnapshot), _, true)
 1083    ).
 1084
 1085
 1086%!  delete_db(+DB)
 1087%
 1088%   Remove snapshot and journal file for DB.
 1089
 1090delete_db(DB) :-
 1091    with_mutex(rdf_journal_file,
 1092               delete_db_(DB)).
 1093
 1094delete_db_(DB) :-
 1095    close_journal_(DB),
 1096    db_abs_files(DB, Snapshot, Journal),
 1097    !,
 1098    (   exists_file(Journal)
 1099    ->  delete_file(Journal)
 1100    ;   true
 1101    ),
 1102    (   exists_file(Snapshot)
 1103    ->  delete_file(Snapshot)
 1104    ;   true
 1105    ).
 1106delete_db_(_).
 1107
 1108                 /*******************************
 1109                 *             LOCKING          *
 1110                 *******************************/
 1111
 1112%!  lock_db(+Dir)
 1113%
 1114%   Lock the database directory Dir.
 1115
 1116lock_db(Dir) :-
 1117    lockfile(Dir, File),
 1118    catch(open(File, update, Out, [lock(write), wait(false)]),
 1119          error(permission_error(Access, _, _), _),
 1120          locked_error(Access, Dir)),
 1121    (   current_prolog_flag(pid, PID)
 1122    ->  true
 1123    ;   PID = 0                     % TBD: Fix in Prolog
 1124    ),
 1125    time_stamp(Now),
 1126    gethostname(Host),
 1127    format(Out, '/* RDF Database is in use */~n~n', []),
 1128    format(Out, '~q.~n', [ locked([ time(Now),
 1129                                    pid(PID),
 1130                                    host(Host)
 1131                                  ])
 1132                         ]),
 1133    flush_output(Out),
 1134    set_end_of_stream(Out),
 1135    assert(rdf_lock(Dir, lock(Out, File))),
 1136    at_halt(unlock_db(Dir)).
 1137
 1138locked_error(lock, Dir) :-
 1139    lockfile(Dir, File),
 1140    (   catch(read_file_to_terms(File, Terms, []), _, fail),
 1141        Terms = [locked(Args)]
 1142    ->  Context = rdf_locked(Args)
 1143    ;   Context = context(_, 'Database is in use')
 1144    ),
 1145    throw(error(permission_error(lock, rdf_db, Dir), Context)).
 1146locked_error(open, Dir) :-
 1147    throw(error(permission_error(lock, rdf_db, Dir),
 1148                context(_, 'Lock file cannot be opened'))).
 1149
 1150%!  unlock_db(+Dir) is det.
 1151%!  unlock_db(+Stream, +File) is det.
 1152
 1153unlock_db(Dir) :-
 1154    retract(rdf_lock(Dir, lock(Out, File))),
 1155    !,
 1156    unlock_db(Out, File).
 1157unlock_db(_).
 1158
 1159unlock_db(Out, File) :-
 1160    close(Out),
 1161    delete_file(File).
 1162
 1163                 /*******************************
 1164                 *           FILENAMES          *
 1165                 *******************************/
 1166
 1167lockfile(Dir, LockFile) :-
 1168    atomic_list_concat([Dir, /, lock], LockFile).
 1169
 1170directory_levels(Levels) :-
 1171    rdf_option(directory_levels(Levels)),
 1172    !.
 1173directory_levels(2).
 1174
 1175db_file(Base, File) :-
 1176    rdf_directory(Dir),
 1177    directory_levels(Levels),
 1178    db_file(Dir, Base, Levels, File).
 1179
 1180db_file(Dir, Base, Levels, File) :-
 1181    dir_levels(Base, Levels, Segments, [Base]),
 1182    atomic_list_concat([Dir|Segments], /, File).
 1183
 1184open_db(Base, Mode, Stream, Options) :-
 1185    db_file(Base, File),
 1186    create_directory_levels(File),
 1187    open(File, Mode, Stream, [encoding(utf8)|Options]).
 1188
 1189create_directory_levels(_File) :-
 1190    rdf_option(directory_levels(0)),
 1191    !.
 1192create_directory_levels(File) :-
 1193    file_directory_name(File, Dir),
 1194    make_directory_path(Dir).
 1195
 1196exists_db(Base) :-
 1197    db_file(Base, File),
 1198    exists_file(File).
 1199
 1200%!  dir_levels(+File, +Levels, ?Segments, ?Tail) is det.
 1201%
 1202%   Create a list of intermediate directory names for File.  Each
 1203%   directory consists of two hexadecimal digits.
 1204
 1205dir_levels(_, 0, Segments, Segments) :- !.
 1206dir_levels(File, Levels, Segments, Tail) :-
 1207    rdf_atom_md5(File, 1, Hash),
 1208    create_dir_levels(Levels, 0, Hash, Segments, Tail).
 1209
 1210create_dir_levels(0, _, _, Segments, Segments) :- !.
 1211create_dir_levels(N, S, Hash, [S1|Segments0], Tail) :-
 1212    sub_atom(Hash, S, 2, _, S1),
 1213    S2 is S+2,
 1214    N2 is N-1,
 1215    create_dir_levels(N2, S2, Hash, Segments0, Tail).
 1216
 1217
 1218%!  db_files(+DB, -Snapshot, -Journal).
 1219%!  db_files(-DB, +Snapshot, -Journal).
 1220%!  db_files(-DB, -Snapshot, +Journal).
 1221%
 1222%   True if named graph DB is represented  by the files Snapshot and
 1223%   Journal. The filenames are local   to the directory representing
 1224%   the store.
 1225
 1226db_files(DB, Snapshot, Journal) :-
 1227    nonvar(DB),
 1228    !,
 1229    rdf_db_to_file(DB, Base),
 1230    atom_concat(Base, '.trp', Snapshot),
 1231    atom_concat(Base, '.jrn', Journal).
 1232db_files(DB, Snapshot, Journal) :-
 1233    nonvar(Snapshot),
 1234    !,
 1235    atom_concat(Base, '.trp', Snapshot),
 1236    atom_concat(Base, '.jrn', Journal),
 1237    rdf_db_to_file(DB, Base).
 1238db_files(DB, Snapshot, Journal) :-
 1239    nonvar(Journal),
 1240    !,
 1241    atom_concat(Base, '.jrn', Journal),
 1242    atom_concat(Base, '.trp', Snapshot),
 1243    rdf_db_to_file(DB, Base).
 1244
 1245db_abs_files(DB, Snapshot, Journal) :-
 1246    db_files(DB, Snapshot0, Journal0),
 1247    db_file(Snapshot0, Snapshot),
 1248    db_file(Journal0, Journal).
 1249
 1250
 1251%!  rdf_journal_file(+Graph, -File) is semidet.
 1252%!  rdf_journal_file(-Graph, -File) is nondet.
 1253%
 1254%   True if File the name of the existing journal file for Graph.
 1255
 1256rdf_journal_file(Graph, Journal) :-
 1257    (   var(Graph)
 1258    ->  rdf_graph(Graph)
 1259    ;   true
 1260    ),
 1261    db_abs_files(Graph, _Snapshot, Journal),
 1262    exists_file(Journal).
 1263
 1264
 1265%!  rdf_snapshot_file(+Graph, -File) is semidet.
 1266%!  rdf_snapshot_file(-Graph, -File) is nondet.
 1267%
 1268%   True if File the name of the existing snapshot file for Graph.
 1269
 1270rdf_snapshot_file(Graph, Snapshot) :-
 1271    (   var(Graph)
 1272    ->  rdf_graph(Graph)    % also pick the empty graphs
 1273    ;   true
 1274    ),
 1275    db_abs_files(Graph, Snapshot, _Journal),
 1276    exists_file(Snapshot).
 1277
 1278
 1279%!  rdf_db_to_file(+DB, -File) is det.
 1280%!  rdf_db_to_file(-DB, +File) is det.
 1281%
 1282%   Translate between database encoding (often an   file or URL) and
 1283%   the name we store in the  directory.   We  keep  a cache for two
 1284%   reasons. Speed, but much more important   is that the mapping of
 1285%   raw --> encoded provided by  www_form_encode/2 is not guaranteed
 1286%   to be unique by the W3C standards.
 1287
 1288rdf_db_to_file(DB, File) :-
 1289    file_base_db(File, DB),
 1290    !.
 1291rdf_db_to_file(DB, File) :-
 1292    url_to_filename(DB, File),
 1293    assert(file_base_db(File, DB)).
 1294
 1295%!  url_to_filename(+URL, -FileName) is det.
 1296%!  url_to_filename(-URL, +FileName) is det.
 1297%
 1298%   Turn  a  valid  URL  into  a  filename.  Earlier  versions  used
 1299%   www_form_encode/2, but this can produce  characters that are not
 1300%   valid  in  filenames.  We  will  use    the   same  encoding  as
 1301%   www_form_encode/2,  but  using  our  own    rules   for  allowed
 1302%   characters. The only requirement is that   we avoid any filename
 1303%   special character in use.  The   current  encoding  use US-ASCII
 1304%   alnum characters, _ and %
 1305
 1306url_to_filename(URL, FileName) :-
 1307    atomic(URL),
 1308    !,
 1309    atom_codes(URL, Codes),
 1310    phrase(url_encode(EncCodes), Codes),
 1311    atom_codes(FileName, EncCodes).
 1312url_to_filename(URL, FileName) :-
 1313    uri_encoded(path, URL, FileName).
 1314
 1315url_encode([0'+|T]) -->
 1316    " ",
 1317    !,
 1318    url_encode(T).
 1319url_encode([C|T]) -->
 1320    alphanum(C),
 1321    !,
 1322    url_encode(T).
 1323url_encode([C|T]) -->
 1324    no_enc_extra(C),
 1325    !,
 1326    url_encode(T).
 1327url_encode(Enc) -->
 1328    (   "\r\n"
 1329    ;   "\n"
 1330    ),
 1331    !,
 1332    { string_codes("%0D%0A", Codes),
 1333      append(Codes, T, Enc)
 1334    },
 1335    url_encode(T).
 1336url_encode([]) -->
 1337    eos,
 1338    !.
 1339url_encode([0'%,D1,D2|T]) -->
 1340    [C],
 1341    { Dv1 is (C>>4 /\ 0xf),
 1342      Dv2 is (C /\ 0xf),
 1343      code_type(D1, xdigit(Dv1)),
 1344      code_type(D2, xdigit(Dv2))
 1345    },
 1346    url_encode(T).
 1347
 1348eos([], []).
 1349
 1350alphanum(C) -->
 1351    [C],
 1352    { C < 128,                      % US-ASCII
 1353      code_type(C, alnum)
 1354    }.
 1355
 1356no_enc_extra(0'_) --> "_".
 1357
 1358
 1359                 /*******************************
 1360                 *             REINDEX          *
 1361                 *******************************/
 1362
 1363%!  reindex_db(+Dir, +Levels)
 1364%
 1365%   Reindex the database by creating intermediate directories.
 1366
 1367reindex_db(Dir, Levels) :-
 1368    directory_files(Dir, Files),
 1369    reindex_files(Files, Dir, '.', 0, Levels),
 1370    remove_empty_directories(Files, Dir).
 1371
 1372reindex_files([], _, _, _, _).
 1373reindex_files([Nofollow|Files], Dir, Prefix, CLevel, Levels) :-
 1374    nofollow(Nofollow),
 1375    !,
 1376    reindex_files(Files, Dir, Prefix, CLevel, Levels).
 1377reindex_files([File|Files], Dir, Prefix, CLevel, Levels) :-
 1378    CLevel \== Levels,
 1379    file_name_extension(_Base, Ext, File),
 1380    db_extension(Ext),
 1381    !,
 1382    directory_file_path(Prefix, File, DBFile),
 1383    directory_file_path(Dir, DBFile, OldPath),
 1384    db_file(Dir, File, Levels, NewPath),
 1385    debug(rdf_persistency, 'Rename ~q --> ~q', [OldPath, NewPath]),
 1386    file_directory_name(NewPath, NewDir),
 1387    make_directory_path(NewDir),
 1388    rename_file(OldPath, NewPath),
 1389    reindex_files(Files, Dir, Prefix, CLevel, Levels).
 1390reindex_files([D|Files], Dir, Prefix, CLevel, Levels) :-
 1391    directory_file_path(Prefix, D, SubD),
 1392    directory_file_path(Dir, SubD, AbsD),
 1393    exists_directory(AbsD),
 1394    \+ read_link(AbsD, _, _),      % Do not follow links
 1395    !,
 1396    directory_files(AbsD, SubFiles),
 1397    CLevel2 is CLevel + 1,
 1398    reindex_files(SubFiles, Dir, SubD, CLevel2, Levels),
 1399    reindex_files(Files, Dir, Prefix, CLevel, Levels).
 1400reindex_files([_|Files], Dir, Prefix, CLevel, Levels) :-
 1401    reindex_files(Files, Dir, Prefix, CLevel, Levels).
 1402
 1403
 1404remove_empty_directories([], _).
 1405remove_empty_directories([File|Files], Dir) :-
 1406    \+ nofollow(File),
 1407    directory_file_path(Dir, File, Path),
 1408    exists_directory(Path),
 1409    \+ read_link(Path, _, _),
 1410    !,
 1411    directory_files(Path, Content),
 1412    exclude(nofollow, Content, RealContent),
 1413    (   RealContent == []
 1414    ->  debug(rdf_persistency, 'Remove empty dir ~q', [Path]),
 1415        delete_directory(Path)
 1416    ;   remove_empty_directories(RealContent, Path)
 1417    ),
 1418    remove_empty_directories(Files, Dir).
 1419remove_empty_directories([_|Files], Dir) :-
 1420    remove_empty_directories(Files, Dir).
 1421
 1422
 1423                 /*******************************
 1424                 *            PREFIXES          *
 1425                 *******************************/
 1426
 1427save_prefixes(Dir) :-
 1428    atomic_list_concat([Dir, /, 'prefixes.db'], PrefixFile),
 1429    setup_call_cleanup(open(PrefixFile, write, Out, [encoding(utf8)]),
 1430                       write_prefixes(Out),
 1431                       close(Out)).
 1432
 1433write_prefixes(Out) :-
 1434    format(Out, '% Snapshot of defined RDF prefixes~n~n', []),
 1435    forall(rdf_current_ns(Alias, URI),
 1436           format(Out, 'prefix(~q, ~q).~n', [Alias, URI])).
 1437
 1438%!  load_prefixes(+RDFDBDir) is det.
 1439%
 1440%   If the file RDFDBDir/prefixes.db exists,  load the prefixes. The
 1441%   prefixes are registered using rdf_register_ns/3. Possible errors
 1442%   because the prefix  definitions  have   changed  are  printed as
 1443%   warnings, retaining the  old  definition.   Note  that  changing
 1444%   prefixes generally requires reloading all RDF from the source.
 1445
 1446load_prefixes(Dir) :-
 1447    atomic_list_concat([Dir, /, 'prefixes.db'], PrefixFile),
 1448    (   exists_file(PrefixFile)
 1449    ->  setup_call_cleanup(open(PrefixFile, read, In, [encoding(utf8)]),
 1450                           read_prefixes(In),
 1451                           close(In))
 1452    ;   true
 1453    ).
 1454
 1455read_prefixes(Stream) :-
 1456    read_term(Stream, T0, []),
 1457    read_prefixes(T0, Stream).
 1458
 1459read_prefixes(end_of_file, _) :- !.
 1460read_prefixes(prefix(Alias, URI), Stream) :-
 1461    !,
 1462    must_be(atom, Alias),
 1463    must_be(atom, URI),
 1464    catch(rdf_register_ns(Alias, URI, []), E,
 1465          print_message(warning, E)),
 1466    read_term(Stream, T, []),
 1467    read_prefixes(T, Stream).
 1468read_prefixes(Term, _) :-
 1469    domain_error(prefix_term, Term).
 1470
 1471
 1472                 /*******************************
 1473                 *              UTIL            *
 1474                 *******************************/
 1475
 1476%!  mkdir(+Directory)
 1477%
 1478%   Create a directory if it does not already exist.
 1479
 1480mkdir(Directory) :-
 1481    exists_directory(Directory),
 1482    !.
 1483mkdir(Directory) :-
 1484    make_directory(Directory).
 1485
 1486%!  time_stamp(-Integer)
 1487%
 1488%   Return time-stamp rounded to integer.
 1489
 1490time_stamp(Int) :-
 1491    get_time(Now),
 1492    Int is round(Now).
 1493
 1494
 1495                 /*******************************
 1496                 *            MESSAGES          *
 1497                 *******************************/
 1498
 1499:- multifile
 1500    prolog:message/3,
 1501    prolog:message_context/3. 1502
 1503prolog:message(rdf(Term)) -->
 1504    message(Term).
 1505
 1506message(restoring(Type, Count, Jobs)) -->
 1507    [ 'Restoring ~D ~w using ~D concurrent workers'-[Count, Type, Jobs] ].
 1508message(restore(attached(Graphs, Triples, Time/Wall))) -->
 1509    { catch(Percent is round(100*Time/Wall), _, Percent = 0) },
 1510    [ 'Loaded ~D graphs (~D triples) in ~2f sec. (~d% CPU = ~2f sec.)'-
 1511      [Graphs, Triples, Wall, Percent, Time] ].
 1512% attach_graph/2
 1513message(restore(true, Action)) -->
 1514    !,
 1515    silent_message(Action).
 1516message(restore(brief, Action)) -->
 1517    !,
 1518    brief_message(Action).
 1519message(restore(_, Graph)) -->
 1520    [ 'Restoring ~p ... '-[Graph], flush ].
 1521message(restore(_, snapshot(_))) -->
 1522    [ at_same_line, '(snapshot) '-[], flush ].
 1523message(restore(_, journal(_))) -->
 1524    [ at_same_line, '(journal) '-[], flush ].
 1525message(restore(_, done(_, Time, Count))) -->
 1526    [ at_same_line, '~D triples in ~2f sec.'-[Count, Time] ].
 1527% load_source/4
 1528message(restore(_, snapshot(G, _))) -->
 1529    [ 'Restoring ~p\t(snapshot)'-[G], flush ].
 1530message(restore(_, journal(G, _))) -->
 1531    [ 'Restoring ~p\t(journal)'-[G], flush ].
 1532message(restore(_, done(_, Time, Count))) -->
 1533    [ at_same_line, '~D triples in ~2f sec.'-[Count, Time] ].
 1534% journal handling
 1535message(update_failed(S,P,O,Action)) -->
 1536    [ 'Failed to update <~p ~p ~p> with ~p'-[S,P,O,Action] ].
 1537% directory reindexing
 1538message(reindex(Count, Depth)) -->
 1539    [ 'Restructuring database with ~d levels (~D graphs)'-[Depth, Count] ].
 1540message(reindex(Depth)) -->
 1541    [ 'Fixing database directory structure (~d levels)'-[Depth] ].
 1542message(read_only) -->
 1543    [ 'Cannot write persistent store; continuing in read-only mode.', nl,
 1544      'All changes to the RDF store will be lost if this process terminates.'
 1545    ].
 1546
 1547silent_message(_Action) --> [].
 1548
 1549brief_message(done(Graph, _Time, _Count, Nth, Total)) -->
 1550    { file_base_name(Graph, Base) },
 1551    [ at_same_line,
 1552      '\r~p~`.t ~D of ~D graphs~72|'-[Base, Nth, Total],
 1553      flush
 1554    ].
 1555brief_message(_) --> [].
 1556
 1557
 1558prolog:message_context(rdf_locked(Args)) -->
 1559    { memberchk(time(Time), Args),
 1560      memberchk(pid(Pid), Args),
 1561      format_time(string(S), '%+', Time)
 1562    },
 1563    [ nl,
 1564      'locked at ~s by process id ~w'-[S,Pid]
 1565    ]